summaryrefslogtreecommitdiff
path: root/servers/hello_world/tests/test_remote.py
diff options
context:
space:
mode:
authorDawid Rycerz <dawid@rycerz.xyz>2025-03-28 20:53:41 +0100
committerDawid Rycerz <dawid@rycerz.xyz>2025-03-28 20:53:41 +0100
commit1745749cd2745c94c3f852e9c02dfde19d8d9c20 (patch)
tree8ed13f3de5fac78d804124e27fbcf1b678e83253 /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.py63
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