diff options
| author | Dawid Rycerz <dawid@rycerz.xyz> | 2025-03-28 20:53:41 +0100 |
|---|---|---|
| committer | Dawid Rycerz <dawid@rycerz.xyz> | 2025-03-28 20:53:41 +0100 |
| commit | 1745749cd2745c94c3f852e9c02dfde19d8d9c20 (patch) | |
| tree | 8ed13f3de5fac78d804124e27fbcf1b678e83253 /servers/hello_world/tests/test_remote.py | |
Fix ruff errors and warnings in hello_world service
Diffstat (limited to 'servers/hello_world/tests/test_remote.py')
| -rw-r--r-- | servers/hello_world/tests/test_remote.py | 63 |
1 files changed, 63 insertions, 0 deletions
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 |
