From 1745749cd2745c94c3f852e9c02dfde19d8d9c20 Mon Sep 17 00:00:00 2001 From: Dawid Rycerz Date: Fri, 28 Mar 2025 20:53:41 +0100 Subject: Fix ruff errors and warnings in hello_world service --- servers/hello_world/tests/conftest.py | 35 ++++++++ servers/hello_world/tests/test_cli.py | 113 ++++++++++++++++++++++++++ servers/hello_world/tests/test_handlers.py | 65 +++++++++++++++ servers/hello_world/tests/test_integration.py | 74 +++++++++++++++++ servers/hello_world/tests/test_prompts.py | 47 +++++++++++ servers/hello_world/tests/test_remote.py | 63 ++++++++++++++ servers/hello_world/tests/test_server.py | 22 +++++ 7 files changed, 419 insertions(+) create mode 100644 servers/hello_world/tests/conftest.py create mode 100644 servers/hello_world/tests/test_cli.py create mode 100644 servers/hello_world/tests/test_handlers.py create mode 100644 servers/hello_world/tests/test_integration.py create mode 100644 servers/hello_world/tests/test_prompts.py create mode 100644 servers/hello_world/tests/test_remote.py create mode 100644 servers/hello_world/tests/test_server.py (limited to 'servers/hello_world/tests') diff --git a/servers/hello_world/tests/conftest.py b/servers/hello_world/tests/conftest.py new file mode 100644 index 0000000..1c62bb5 --- /dev/null +++ b/servers/hello_world/tests/conftest.py @@ -0,0 +1,35 @@ +import pytest +import sys +import os +from unittest.mock import AsyncMock, MagicMock, patch + +# Add the src directory to the Python path for testing +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../src'))) + +# Mock the mcp.server.remote module +mock_remote = MagicMock() +mock_remote.serve_remote = AsyncMock() +sys.modules['mcp.server.remote'] = mock_remote + +@pytest.fixture +def hello_world_server(): + """Create a HelloWorldServer instance for testing.""" + from mcp_server_hello_world.server import HelloWorldServer + return HelloWorldServer() + +@pytest.fixture +def mcp_server_mock(): + """Create a mock MCP Server instance with all handlers.""" + server = MagicMock() + server.list_resources = MagicMock(return_value=lambda: None) + server.read_resource = MagicMock(return_value=lambda: None) + server.list_prompts = MagicMock(return_value=lambda: None) + server.get_prompt = MagicMock(return_value=lambda: None) + server.list_tools = MagicMock(return_value=lambda: None) + server.call_tool = MagicMock(return_value=lambda: None) + server.run = AsyncMock() + server.get_capabilities = MagicMock(return_value={}) + server.request_context = MagicMock() + server.request_context.session = MagicMock() + server.request_context.session.send_resource_updated = AsyncMock() + return server diff --git a/servers/hello_world/tests/test_cli.py b/servers/hello_world/tests/test_cli.py new file mode 100644 index 0000000..48fa89c --- /dev/null +++ b/servers/hello_world/tests/test_cli.py @@ -0,0 +1,113 @@ +import pytest +import sys +from unittest.mock import patch, MagicMock +import argparse +from mcp_server_hello_world.cli import run_server, parse_args, validate_args + +def test_cli_argument_parsing(): + """Test the argument parsing in the CLI module.""" + # Test default arguments + with patch('argparse.ArgumentParser.parse_args') as mock_parse_args: + + # Set up the mock to return default values + mock_args = argparse.Namespace( + transport="stdio", + host="0.0.0.0", + port=8080, + log_level="INFO" + ) + mock_parse_args.return_value = mock_args + + # Call the parse_args function + args = parse_args() + + # Check the arguments + assert args.transport == "stdio" + assert args.host == "0.0.0.0" + assert args.port == 8080 + assert args.log_level == "INFO" + + # Test custom arguments + with patch('argparse.ArgumentParser.parse_args') as mock_parse_args: + + # Set up the mock to return custom values + mock_args = argparse.Namespace( + transport="remote", + host="127.0.0.1", + port=9090, + log_level="DEBUG" + ) + mock_parse_args.return_value = mock_args + + # Call the parse_args function + args = parse_args() + + # Check the arguments + assert args.transport == "remote" + assert args.host == "127.0.0.1" + assert args.port == 9090 + assert args.log_level == "DEBUG" + +def test_run_server(): + """Test the run_server function.""" + with patch('mcp_server_hello_world.cli.parse_args') as mock_parse_args, \ + patch('mcp_server_hello_world.cli.validate_args', return_value=mock_parse_args.return_value) as mock_validate_args, \ + patch('mcp_server_hello_world.cli.setup_logging') as mock_setup_logging, \ + patch('asyncio.run') as mock_run: + + # Set up the mock to return default values + mock_args = argparse.Namespace( + transport="stdio", + host="0.0.0.0", + port=8080, + log_level="INFO" + ) + mock_parse_args.return_value = mock_args + + # Call the run_server function + run_server() + + # Check that the functions were called with the correct arguments + mock_parse_args.assert_called_once() + mock_validate_args.assert_called_once_with(mock_args) + # The setup_logging function is called with the log_level attribute from the args + mock_setup_logging.assert_called_once() + mock_run.assert_called_once() + +def test_argument_validation(): + """Test that the argument parser validates arguments correctly.""" + # Test valid transport values + with patch('sys.argv', ['mcp-server-hello-world', '--transport', 'stdio']): + parser = argparse.ArgumentParser() + parser.add_argument('--transport', choices=["stdio", "remote"]) + args = parser.parse_args() + assert args.transport == "stdio" + + with patch('sys.argv', ['mcp-server-hello-world', '--transport', 'remote']): + parser = argparse.ArgumentParser() + parser.add_argument('--transport', choices=["stdio", "remote"]) + args = parser.parse_args() + assert args.transport == "remote" + + # Test invalid transport value + with patch('sys.argv', ['mcp-server-hello-world', '--transport', 'invalid']), \ + patch('sys.stderr', MagicMock()), \ + pytest.raises(SystemExit): + parser = argparse.ArgumentParser() + parser.add_argument('--transport', choices=["stdio", "remote"]) + parser.parse_args() + + # Test port validation + with patch('sys.argv', ['mcp-server-hello-world', '--port', '8080']): + parser = argparse.ArgumentParser() + parser.add_argument('--port', type=int) + args = parser.parse_args() + assert args.port == 8080 + + # Test invalid port (non-integer) + with patch('sys.argv', ['mcp-server-hello-world', '--port', 'invalid']), \ + patch('sys.stderr', MagicMock()), \ + pytest.raises(SystemExit): + parser = argparse.ArgumentParser() + parser.add_argument('--port', type=int) + parser.parse_args() diff --git a/servers/hello_world/tests/test_handlers.py b/servers/hello_world/tests/test_handlers.py new file mode 100644 index 0000000..0c0a45d --- /dev/null +++ b/servers/hello_world/tests/test_handlers.py @@ -0,0 +1,65 @@ +import pytest +import asyncio +import sys +from unittest.mock import AsyncMock, MagicMock, patch +from pydantic import AnyUrl +import mcp.types as types +from mcp_server_hello_world.server import HelloWorldServer, create_server + +class TestHandlers: + """Test the MCP server handlers.""" + + @pytest.mark.asyncio + async def test_resource_handler(self): + """Test the resource handler.""" + # Create a server + mcp = create_server() + + # Get the resource URI + uri = "hello://welcome" + + # Call the resource handler + result = await mcp.read_resource(uri) + + # Verify the result + assert len(result) == 1 + assert result[0].content == "Welcome to the Hello World MCP Server! This is a simple static resource." + + @pytest.mark.asyncio + async def test_tool_handler(self): + """Test the tool handler.""" + # Create a server + mcp = create_server() + + # Call the tool handler + result = await mcp.call_tool("hello", {"name": "Test User"}) + + # Verify the result + assert len(result) == 1 + assert result[0].text == "Hello, Test User! Welcome to the Hello World MCP Server." + + # Test with missing name argument + with pytest.raises(Exception) as excinfo: + await mcp.call_tool("hello", {}) + assert "Error executing tool hello" in str(excinfo.value) + assert "Field required" in str(excinfo.value) + + # Test with unknown tool + with pytest.raises(Exception) as excinfo: + await mcp.call_tool("unknown", {}) + assert "Unknown tool: unknown" in str(excinfo.value) + + @pytest.mark.asyncio + async def test_prompt_handler(self): + """Test the prompt handler.""" + # Create a server + mcp = create_server() + + # Call the prompt handler + result = await mcp.get_prompt("greeting", {"name": "Test User"}) + + # Verify the result + assert len(result.messages) > 0 + assert result.messages[0].role == "user" + assert result.messages[0].content.type == "text" + assert "Hello there! You've chosen to use the greeting prompt with the name: Test User." in result.messages[0].content.text diff --git a/servers/hello_world/tests/test_integration.py b/servers/hello_world/tests/test_integration.py new file mode 100644 index 0000000..5948dae --- /dev/null +++ b/servers/hello_world/tests/test_integration.py @@ -0,0 +1,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 diff --git a/servers/hello_world/tests/test_prompts.py b/servers/hello_world/tests/test_prompts.py new file mode 100644 index 0000000..9709718 --- /dev/null +++ b/servers/hello_world/tests/test_prompts.py @@ -0,0 +1,47 @@ +import pytest +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch +import mcp.types as types +from mcp_server_hello_world.server import HelloWorldServer, create_server, GREETING_PROMPT_TEMPLATE + +class TestPrompts: + """Test the prompt functionality.""" + + @pytest.mark.asyncio + async def test_list_prompts(self): + """Test the list_prompts method.""" + # Create a server + mcp = create_server() + + # Get the list of prompts + prompts = await mcp.list_prompts() + + # Verify the result + assert len(prompts) == 1 + assert prompts[0].name == "greeting" + assert "simple greeting prompt" in prompts[0].description.lower() + + @pytest.mark.asyncio + async def test_get_prompt(self): + """Test the get_prompt method.""" + # Create a server + mcp = create_server() + + # Get the prompt + prompt = await mcp.get_prompt("greeting", {"name": "Test User"}) + + # Verify the result + expected_prompt = GREETING_PROMPT_TEMPLATE.format(name="Test User") + assert len(prompt.messages) > 0 + assert prompt.messages[0].role == "user" + assert prompt.messages[0].content.type == "text" + assert expected_prompt.strip() in prompt.messages[0].content.text + + # Test with unknown prompt + with pytest.raises(ValueError): + await mcp.get_prompt("unknown", {"name": "Test User"}) + + # Test with missing name argument + with pytest.raises(ValueError) as excinfo: + await mcp.get_prompt("greeting", {}) + assert "Missing required arguments" in str(excinfo.value) diff --git a/servers/hello_world/tests/test_remote.py b/servers/hello_world/tests/test_remote.py new file mode 100644 index 0000000..c98d129 --- /dev/null +++ b/servers/hello_world/tests/test_remote.py @@ -0,0 +1,63 @@ +import pytest +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch +import sys + +class TestTransport: + """Test the transport functionality.""" + + @pytest.mark.asyncio + async def test_transport_selection(self): + """Test that the server selects the correct transport based on the argument.""" + # 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", "0.0.0.0", 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_remote_transport(self): + """Test that the server can be started with remote transport.""" + # Import the main function here to avoid issues with the mocking + from mcp_server_hello_world.server import main + + # Mock the run_sse_async method + with patch('mcp.server.fastmcp.FastMCP.run_sse_async') as mock_run_sse: + + # Set up the mocks + mock_run_sse.return_value = AsyncMock() + + # Start the server with remote transport + task = asyncio.create_task(main("remote", "0.0.0.0", 8080)) + + # Wait for the server to start + await asyncio.sleep(0.1) + + # Check that run_sse_async was called without parameters + mock_run_sse.assert_called_once_with() + + # Cancel the task + task.cancel() + try: + await task + except asyncio.CancelledError: + pass diff --git a/servers/hello_world/tests/test_server.py b/servers/hello_world/tests/test_server.py new file mode 100644 index 0000000..13b8f90 --- /dev/null +++ b/servers/hello_world/tests/test_server.py @@ -0,0 +1,22 @@ +import pytest +from mcp_server_hello_world.server import HelloWorldServer + +class TestHelloWorldServer: + """Test the HelloWorldServer class functionality.""" + + def test_init(self): + """Test that the server initializes with the correct welcome message.""" + server = HelloWorldServer() + assert server.welcome_message == "Welcome to the Hello World MCP Server! This is a simple static resource." + + def test_get_greeting(self): + """Test that the get_greeting method returns the correct greeting.""" + server = HelloWorldServer() + name = "Test User" + expected_greeting = f"Hello, {name}! Welcome to the Hello World MCP Server." + assert server.get_greeting(name) == expected_greeting + + # Test with different name + another_name = "Another User" + expected_greeting = f"Hello, {another_name}! Welcome to the Hello World MCP Server." + assert server.get_greeting(another_name) == expected_greeting -- cgit v1.2.3