Add test demonstrating Windows cleanup issue #1027

This test shows that cleanup code after yield in MCP server lifespan
is not executed on Windows when the process is terminated with
process.terminate().

The test includes two scenarios:
1. Current behavior - cleanup code is unreachable on Windows
2. Proposed solution - using stdin closure for graceful shutdown

The tests use pytest.xfail to mark expected failures on Windows.

Github-Issue:#1027

Author: felixweinberger

tests/issues/test_1027_win_unreachable_cleanup.py

@@ -0,0 +1,261 @@
+"""
+Test for issue #1027: The cleanup procedure after "yield" in lifespan is unreachable on Windows
+
+This test demonstrates that on Windows, when an MCP server is terminated via process.terminate(),
+the cleanup code after yield in the lifespan context manager is not executed.
+
+The issue occurs because Windows' TerminateProcess() forcefully kills the process without
+allowing cleanup handlers to run.
+"""
+
+import asyncio
+import sys
+import tempfile
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+
+@pytest.mark.anyio
+async def test_lifespan_cleanup_executed():
+    """
+    Test that verifies cleanup code in MCP server lifespan is executed.
+    
+    This test creates an MCP server that writes to marker files:
+    1. When the server starts (before yield)
+    2. When the server cleanup runs (after yield)
+    
+    On Windows with the current implementation, the cleanup file is never created
+    because process.terminate() kills the process immediately.
+    """
+    
+    # Create marker files to track server lifecycle
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f:
+        startup_marker = f.name
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f:
+        cleanup_marker = f.name
+    
+    # Remove the files so we can detect when they're created
+    Path(startup_marker).unlink()
+    Path(cleanup_marker).unlink()
+    
+    # Create a minimal MCP server that tracks lifecycle
+    server_code = textwrap.dedent(f"""
+import asyncio
+import sys
+from pathlib import Path
+from contextlib import asynccontextmanager
+from mcp.server.lowlevel import Server
+from mcp.server.stdio import stdio_server
+
+STARTUP_MARKER = {repr(startup_marker)}
+CLEANUP_MARKER = {repr(cleanup_marker)}
+
+@asynccontextmanager
+async def lifespan(server):
+    # Write startup marker
+    Path(STARTUP_MARKER).write_text("started")
+    try:
+        yield
+    finally:
+        # This cleanup code should run when server shuts down
+        Path(CLEANUP_MARKER).write_text("cleaned up")
+
+async def main():
+    async with stdio_server() as (read, write):
+        server = Server("test-server", version="1.0.0")
+        server.set_lifespan(lifespan)
+        
+        async def handle_initialize(params):
+            return {{"protocolVersion": "1.0.0", "capabilities": {{}}}}
+        
+        server.set_request_handler("initialize", handle_initialize)
+        
+        async def handle_echo(params):
+            return {{"text": params.get("text", "")}}
+        
+        server.set_request_handler("test/echo", handle_echo)
+        
+        await server.run(read, write)
+
+if __name__ == "__main__":
+    import asyncio
+    asyncio.run(main())
+""")
+    
+    # Write the server script to a temporary file
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".py") as f:
+        server_script = f.name
+        f.write(server_code)
+    
+    try:
+        # Launch the MCP server
+        params = StdioServerParameters(
+            command=sys.executable,
+            args=[server_script]
+        )
+        
+        async with stdio_client(params) as (read, write):
+            async with ClientSession(read, write) as session:
+                # Initialize the session
+                result = await session.initialize()
+                assert result.protocolVersion == "1.0.0"
+                
+                # Verify startup marker was created
+                assert Path(startup_marker).exists(), "Server startup marker not created"
+                assert Path(startup_marker).read_text() == "started"
+                
+                # Make a test request to ensure server is working
+                response = await session.send_request("test/echo", {"text": "hello"})
+                assert response.text == "hello"
+                
+                # Session will be closed when exiting the context manager
+        
+        # Give server a moment to run cleanup (if it can)
+        await asyncio.sleep(0.5)
+        
+        # Check if cleanup marker was created
+        if sys.platform == "win32":
+            # On Windows, this assertion currently fails because cleanup doesn't run
+            # When issue #1027 is fixed, this should pass
+            pytest.xfail(
+                "Cleanup code after yield is not executed on Windows (issue #1027)"
+                if not Path(cleanup_marker).exists()
+                else "Cleanup unexpectedly succeeded - issue may be fixed!"
+            )
+        else:
+            # On Unix systems, cleanup should work
+            assert Path(cleanup_marker).exists(), "Server cleanup marker not created"
+            assert Path(cleanup_marker).read_text() == "cleaned up"
+    
+    finally:
+        # Clean up files
+        for path in [server_script, startup_marker, cleanup_marker]:
+            try:
+                Path(path).unlink()
+            except FileNotFoundError:
+                pass
+
+
+@pytest.mark.anyio
+async def test_stdin_close_triggers_cleanup():
+    """
+    Test that verifies if closing stdin allows cleanup to run.
+    
+    This is the proposed solution from PR #1044 - close stdin first
+    and wait for the server to exit gracefully before terminating.
+    """
+    
+    # Create marker files to track server lifecycle  
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f:
+        startup_marker = f.name
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f:
+        cleanup_marker = f.name
+    
+    # Remove the files so we can detect when they're created
+    Path(startup_marker).unlink()
+    Path(cleanup_marker).unlink()
+    
+    # Create an MCP server that handles stdin closure gracefully
+    server_code = textwrap.dedent(f"""
+import asyncio
+import sys
+from pathlib import Path
+from contextlib import asynccontextmanager
+from mcp.server.lowlevel import Server
+from mcp.server.stdio import stdio_server
+
+STARTUP_MARKER = {repr(startup_marker)}
+CLEANUP_MARKER = {repr(cleanup_marker)}
+
+@asynccontextmanager
+async def lifespan(server):
+    # Write startup marker
+    Path(STARTUP_MARKER).write_text("started")
+    try:
+        yield
+    finally:
+        # This cleanup code should run when stdin closes
+        Path(CLEANUP_MARKER).write_text("cleaned up")
+
+async def main():
+    try:
+        async with stdio_server() as (read, write):
+            server = Server("test-server", version="1.0.0")
+            server.set_lifespan(lifespan)
+            
+            async def handle_initialize(params):
+                return {{"protocolVersion": "1.0.0", "capabilities": {{}}}}
+            
+            server.set_request_handler("initialize", handle_initialize)
+            
+            # The server should exit gracefully when stdin closes
+            await server.run(read, write)
+    except Exception:
+        # Server might get EOF or other errors when stdin closes
+        pass
+
+if __name__ == "__main__":
+    asyncio.run(main())
+""")
+    
+    # Write the server script to a temporary file
+    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".py") as f:
+        server_script = f.name
+        f.write(server_code)
+    
+    try:
+        # This test manually manages the process to test stdin closure
+        import anyio
+        from mcp.client.stdio import _create_platform_compatible_process
+        
+        # Start the server process
+        process = await _create_platform_compatible_process(
+            command=sys.executable,
+            args=[server_script],
+            env=None,
+            stderr=None,
+            cwd=None
+        )
+        
+        # Wait for server to start
+        await asyncio.sleep(0.5)
+        assert Path(startup_marker).exists(), "Server startup marker not created"
+        
+        # Close stdin to signal shutdown
+        if process.stdin:
+            await process.stdin.aclose()
+        
+        # Wait for process to exit gracefully
+        try:
+            with anyio.fail_after(2.0):
+                await process.wait()
+        except TimeoutError:
+            # If it doesn't exit after stdin close, terminate it
+            process.terminate()
+            await process.wait()
+        
+        # Check if cleanup ran
+        await asyncio.sleep(0.5)
+        
+        # This should work if the server properly handles stdin closure
+        if Path(cleanup_marker).exists():
+            assert Path(cleanup_marker).read_text() == "cleaned up"
+            # If this works, it shows stdin closure can trigger graceful shutdown
+        else:
+            pytest.xfail(
+                "Server did not run cleanup after stdin closure - "
+                "may need additional server-side handling"
+            )
+    
+    finally:
+        # Clean up files
+        for path in [server_script, startup_marker, cleanup_marker]:
+            try:
+                Path(path).unlink()
+            except FileNotFoundError:
+                pass