brandon/nextcloud-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: Cleanup testing fixtures regarding canceled scopes

Browse Source
This commit is contained in:
Chris Coutinho
2025-10-18 20:11:07 +02:00
parent 371d0c93a5
commit c3ff92a8c1
2 changed files with 425 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/testing-client-sessions-architecture.md
+317
View File
@@ -0,0 +1,317 @@
# Testing Client Sessions Architecture
## Overview
This document compares different approaches to managing MCP client sessions in integration tests, addressing the fundamental incompatibility between pytest-asyncio's fixture management and anyio's structured concurrency requirements.
## The Problem
When using pytest-asyncio with anyio-based libraries (like the MCP Python SDK), session-scoped async generator fixtures encounter a fundamental issue:
1. **pytest-asyncio** runs fixture teardown in a **new asyncio task** using `runner.run()`
2. **anyio** requires that cancel scopes be entered and exited in the **same task**
3. This causes `RuntimeError: Attempted to exit cancel scope in a different task than it was entered in`
This is a **known limitation** documented in the anyio project and is not a bug in either pytest-asyncio or anyio, but rather an inherent incompatibility between their design philosophies.
## Solution Comparison
### Solution 1: Native Async Context Managers with Surgical Exception Handling ✅ **IMPLEMENTED**
**Approach**: Use native `async with` statements for clean code structure, but add targeted exception handling at the pytest fixture level to handle the expected teardown errors.
**Implementation**:
```python
async def create_mcp_client_session(
url: str,
token: str | None = None,
client_name: str = "MCP",
) -> AsyncGenerator[ClientSession, Any]:
"""Uses native async context managers for clean LIFO cleanup."""
headers = {"Authorization": f"Bearer {token}"} if token else None
async with streamablehttp_client(url, headers=headers) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
yield session
@pytest.fixture(scope="session")
async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
"""Fixture with surgical exception handling for pytest-asyncio incompatibility."""
try:
async for session in create_mcp_client_session(
url="http://127.0.0.1:8000/mcp", client_name="Basic MCP"
):
yield session
except RuntimeError as e:
# Only catch the specific expected error during pytest teardown
if "cancel scope" in str(e) and "different task" in str(e):
logger.debug(f"Ignoring expected pytest-asyncio teardown issue: {e}")
else:
# Unexpected RuntimeError - re-raise to fail the test
raise
```
**Pros**:
- ✅ Clean, idiomatic code using native Python context managers
- ✅ Exception handling is surgical - only catches the specific expected error
- ✅ Unexpected errors still propagate and fail tests
- ✅ Can use session-scoped fixtures for performance
- ✅ Easy to understand and maintain
- ✅ Minimal code changes from original implementation
- ✅ No external dependencies required
**Cons**:
- ⚠️ Still requires exception suppression (though targeted)
- ⚠️ String-based exception matching is somewhat fragile
- ⚠️ Must apply the pattern to each session-scoped fixture
- ⚠️ Doesn't solve the root cause
**Verdict**: **Recommended** - Best balance of code clarity, maintainability, and pragmatism.
---
### Solution 2: Task-Isolated Fixtures
**Approach**: Run each fixture's client session in an isolated anyio task group, allowing independent cleanup without cross-fixture interference.
**Implementation**:
```python
@pytest.fixture(scope="session")
async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
"""Fixture with task isolation for clean teardown."""
import anyio
session_holder = {"session": None}
async def create_and_hold_session():
"""Runs in isolated task - creates session and keeps it alive."""
async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
session_holder["session"] = session
# Keep session alive until cancelled
try:
await anyio.sleep_forever()
except anyio.get_cancelled_exc_class():
pass # Expected cancellation
async with anyio.create_task_group() as tg:
tg.start_soon(create_and_hold_session)
# Wait for session to be ready
while session_holder["session"] is None:
await anyio.sleep(0.1)
yield session_holder["session"]
# Task group cancellation ensures clean LIFO cleanup
tg.cancel_scope.cancel()
```
**Pros**:
- ✅ No exception suppression needed
- ✅ Each fixture has its own isolated task scope
- ✅ More theoretically correct approach
- ✅ Can use session-scoped fixtures
**Cons**:
- ❌ Significantly more complex code
- ❌ Harder to understand for developers unfamiliar with anyio
- ❌ Requires understanding of task groups and cancel scopes
- ❌ More boilerplate per fixture
- ❌ Still doesn't solve the fundamental pytest-asyncio incompatibility
- ❌ Polling for session readiness is inelegant
- ❌ Higher cognitive overhead for maintenance
**Verdict**: **Not Recommended** - Complexity outweighs benefits. Consider only if exception handling is completely unacceptable.
---
### Solution 3: Function-Scoped Fixtures with Nested Context Managers
**Approach**: Change fixtures to function scope and rely on Python's context manager nesting for guaranteed LIFO cleanup.
**Implementation**:
```python
@pytest.fixture(scope="function") # Changed from session
async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
"""Function-scoped fixture with natural LIFO cleanup."""
async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
yield session
# For tests needing multiple clients:
@pytest.fixture(scope="function")
async def multi_mcp_clients() -> AsyncGenerator[tuple[ClientSession, ClientSession], Any]:
"""Multiple clients with guaranteed LIFO cleanup through nesting."""
async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (read1, write1, _):
async with ClientSession(read1, write1) as session1:
await session1.initialize()
async with streamablehttp_client("http://127.0.0.1:8001/mcp") as (read2, write2, _):
async with ClientSession(read2, write2) as session2:
await session2.initialize()
yield session1, session2
# Cleanup: session2 -> stream2 -> session1 -> stream1 (LIFO guaranteed)
```
**Pros**:
- ✅ No exception handling needed
- ✅ Simplest to understand
- ✅ Natural LIFO cleanup through Python's context managers
- ✅ Each test gets fresh clients (better isolation)
- ✅ No workarounds or hacks required
**Cons**:
- ❌ Significantly slower tests (new clients per test)
- ❌ Cannot share client state across tests
- ❌ More resource intensive
- ❌ Higher overhead for test suite execution
- ❌ May not be practical for expensive fixtures (e.g., OAuth tokens)
- ❌ Nested context managers become unwieldy with many clients
**Verdict**: **Good Alternative** - Consider for specific fixtures where session scope isn't critical, or for new test files where performance isn't a concern.
---
### Solution 4: Use pytest-trio Instead of pytest-asyncio (Future)
**Approach**: Replace pytest-asyncio with pytest-trio, which was designed with structured concurrency in mind.
**Implementation**:
```python
# pyproject.toml
[tool.pytest.ini_options]
# Remove: asyncio_mode = "auto"
# Add: trio_mode = "auto"
# Fixtures work naturally with trio
@pytest.fixture(scope="session")
async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
yield session
```
**Pros**:
- ✅ No workarounds needed
- ✅ Designed for structured concurrency
- ✅ Theoretically cleanest solution
- ✅ Can use session-scoped fixtures naturally
**Cons**:
- ❌ Requires switching from asyncio to trio backend
- ❌ Major refactoring required
- ❌ May break existing code that assumes asyncio
- ❌ Dependency changes throughout project
- ❌ Team needs to learn trio ecosystem
- ❌ Less ecosystem support than asyncio
**Verdict**: **Not Practical** - Too disruptive for existing projects. Consider only for greenfield projects or major rewrites.
---
## Decision Matrix
| Solution | Code Clarity | Maintenance | Performance | Safety | Effort |
|----------|--------------|-------------|-------------|--------|--------|
| **Solution 1** (Implemented) | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| Solution 2 (Task-Isolated) | ⭐⭐ | ⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐ |
| Solution 3 (Function-Scoped) | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| Solution 4 (pytest-trio) | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐ |
## Implementation Details
### What Changed in Solution 1
1. **`create_mcp_client_session` function** (conftest.py:61-110):
- Replaced manual `__aenter__`/`__aexit__` calls with native `async with` statements
- Removed blanket exception suppression from cleanup logic
- Added clear documentation about LIFO cleanup order
- Simplified from ~60 lines to ~40 lines
2. **Session-scoped MCP client fixtures** (conftest.py:148-1269):
- Added targeted exception handling wrapper
- Only catches specific "cancel scope" + "different task" RuntimeError
- All other exceptions propagate normally
- Applied to: `nc_mcp_client`, `nc_mcp_oauth_client`, `alice_mcp_client`, `bob_mcp_client`, `charlie_mcp_client`, `diana_mcp_client`
3. **Documentation**:
- Added comprehensive docstrings explaining the workaround
- Referenced MCP SDK issue #577 for context
- Documented why this is necessary and not a bug
### Benefits of This Implementation
1. **Clean Core Logic**: The `create_mcp_client_session` function is now clean, idiomatic Python with no workarounds
2. **Isolated Workaround**: Exception handling is confined to pytest fixture level where the issue actually occurs
3. **Surgical Exception Handling**: Only catches the specific expected error, not all RuntimeErrors
4. **Performance**: Maintains session-scoped fixtures for fast test execution
5. **Maintainability**: Easy to understand and modify
6. **Safety**: Real errors still cause test failures
## Testing Results
All tests pass cleanly with the implementation:
```bash
$ uv run pytest tests/server/test_mcp.py -v
============================================= test session starts ==============================================
tests/server/test_mcp.py::test_mcp_connectivity PASSED [ 16%]
tests/server/test_mcp.py::test_mcp_notes_crud_workflow PASSED [ 33%]
tests/server/test_mcp.py::test_mcp_notes_etag_conflict PASSED [ 50%]
tests/server/test_mcp.py::test_mcp_webdav_workflow PASSED [ 66%]
tests/server/test_mcp.py::test_mcp_resources_access PASSED [ 83%]
tests/server/test_mcp.py::test_mcp_calendar_workflow PASSED [100%]
============================================== 6 passed in 39.52s ==============================================
```
## Recommendations
### For This Project: Solution 1 ✅
The implemented solution (Solution 1) is the best fit because:
- Minimal disruption to existing tests
- Clean, maintainable code
- Good performance with session-scoped fixtures
- Targeted exception handling that doesn't hide real errors
### For New Test Files: Consider Solution 3
For new test files where performance isn't critical, consider using function-scoped fixtures (Solution 3):
- No workarounds needed
- Perfect code clarity
- Better test isolation
### For Greenfield Projects: Consider Solution 4
For new projects starting from scratch, consider pytest-trio instead of pytest-asyncio:
- Native structured concurrency support
- No workarounds needed
- Better alignment with modern async Python patterns
## Related Resources
- [MCP Python SDK Issue #577](https://github.com/modelcontextprotocol/python-sdk/issues/577) - Original issue report
- [Anyio Issue #345](https://github.com/agronholm/anyio/issues/345) - Discussion of fixture limitations
- [Nextcloud MCP Note 378555](nextcloud://notes/378555) - Detailed investigation notes
- pytest-asyncio documentation: https://pytest-asyncio.readthedocs.io/
- anyio structured concurrency guide: https://anyio.readthedocs.io/en/stable/basics.html
## Appendix: Why Can't This Be Fixed Upstream?
The incompatibility cannot be "fixed" in either pytest-asyncio or anyio without breaking their core design:
1. **pytest-asyncio** needs to manage fixture lifecycle across different scopes, requiring separate task creation for cleanup
2. **anyio** enforces structured concurrency guarantees by requiring same-task cancel scope entry/exit
3. These requirements are fundamentally incompatible
The maintainers of both projects are aware of this issue, and it's considered an acceptable trade-off given their respective design goals. The recommended approach is to handle it at the application level, as we've done here.
tests/conftest.py
+108 -68
View File
@@ -66,10 +66,14 @@ async def create_mcp_client_session(
"""
Factory function to create an MCP client session with proper lifecycle management.
Uses native async context managers to ensure correct LIFO cleanup order,
eliminating the need for exception suppression. Python's context manager protocol
guarantees that cleanup happens in reverse order of entry.
Consolidates the common pattern used by all MCP client fixtures:
- Creates streamable HTTP client with optional OAuth token
- Initializes MCP ClientSession
- Handles cleanup with proper exception handling
- Ensures proper cleanup without suppressing errors
Args:
url: MCP server URL (e.g., "http://127.0.0.1:8000/mcp")
@@ -78,48 +82,32 @@ async def create_mcp_client_session(
Yields:
Initialized MCP ClientSession
Note:
This implementation uses native async context managers instead of manually
calling __aenter__/__aexit__. This ensures that anyio's structured concurrency
requirements are met, as Python guarantees LIFO cleanup order for nested
context managers. See: https://github.com/modelcontextprotocol/python-sdk/issues/577
"""
logger.info(f"Creating Streamable HTTP client for {client_name}")
# Prepare headers with OAuth token if provided
headers = {"Authorization": f"Bearer {token}"} if token else None
streamable_context = streamablehttp_client(url, headers=headers)
session_context = None
try:
read_stream, write_stream, _ = await streamable_context.__aenter__()
session_context = ClientSession(read_stream, write_stream)
session = await session_context.__aenter__()
await session.initialize()
logger.info(f"{client_name} client session initialized successfully")
# Use native async with - Python ensures LIFO cleanup
# Cleanup order will be: ClientSession.__aexit__ -> streamablehttp_client.__aexit__
async with streamablehttp_client(url, headers=headers) as (
read_stream,
write_stream,
_,
):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
logger.info(f"{client_name} client session initialized successfully")
yield session
yield session
finally:
# Clean up in reverse order, ignoring task scope issues
# See: https://github.com/modelcontextprotocol/python-sdk/issues/577
if session_context is not None:
try:
await session_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(f"Error closing {client_name} session: {e}")
except Exception as e:
logger.warning(f"Error closing {client_name} session: {e}")
try:
await streamable_context.__aexit__(None, None, None)
except RuntimeError as e:
if "cancel scope" in str(e):
logger.debug(f"Ignoring cancel scope teardown issue: {e}")
else:
logger.warning(
f"Error closing {client_name} streamable HTTP client: {e}"
)
except Exception as e:
logger.warning(f"Error closing {client_name} streamable HTTP client: {e}")
# Cleanup happens automatically in LIFO order - no exception suppression needed
logger.debug(f"{client_name} client session cleaned up successfully")
@pytest.fixture(scope="session")
@@ -161,11 +149,28 @@ async def nc_client() -> AsyncGenerator[NextcloudClient, Any]:
async def nc_mcp_client() -> AsyncGenerator[ClientSession, Any]:
"""
Fixture to create an MCP client session for integration tests using streamable-http.
Note: This fixture uses a workaround for pytest-asyncio + anyio incompatibility.
pytest-asyncio runs fixture teardown in a new asyncio task, which violates anyio's
requirement that cancel scopes must be entered/exited in the same task. We catch
and ignore these expected teardown errors while allowing real errors to propagate.
See: https://github.com/modelcontextprotocol/python-sdk/issues/577
"""
async for session in create_mcp_client_session(
url="http://127.0.0.1:8000/mcp", client_name="Basic MCP"
):
yield session
try:
async for session in create_mcp_client_session(
url="http://127.0.0.1:8000/mcp", client_name="Basic MCP"
):
yield session
except RuntimeError as e:
# Expected error during pytest-asyncio fixture teardown
# pytest-asyncio creates a new task for teardown, causing:
# "Attempted to exit cancel scope in a different task than it was entered in"
if "cancel scope" in str(e) and "different task" in str(e):
logger.debug(f"Ignoring expected pytest-asyncio teardown issue: {e}")
else:
# Unexpected RuntimeError - re-raise
raise
@pytest.fixture(scope="session")
@@ -177,13 +182,22 @@ async def nc_mcp_oauth_client(
Connects to the OAuth-enabled MCP server on port 8001 with OAuth authentication.
Uses headless browser automation suitable for CI/CD.
Note: Includes workaround for pytest-asyncio + anyio incompatibility.
See nc_mcp_client fixture for details.
"""
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=playwright_oauth_token,
client_name="OAuth MCP (Playwright)",
):
yield session
try:
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=playwright_oauth_token,
client_name="OAuth MCP (Playwright)",
):
yield session
except RuntimeError as e:
if "cancel scope" in str(e) and "different task" in str(e):
logger.debug(f"Ignoring expected pytest-asyncio teardown issue: {e}")
else:
raise
@pytest.fixture
@@ -1186,21 +1200,35 @@ async def alice_mcp_client(
alice_oauth_token: str,
) -> AsyncGenerator[ClientSession, Any]:
"""MCP client authenticated as alice (owner role)."""
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=alice_oauth_token,
client_name="Alice MCP",
):
yield session
try:
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=alice_oauth_token,
client_name="Alice MCP",
):
yield session
except RuntimeError as e:
if "cancel scope" in str(e) and "different task" in str(e):
logger.debug(f"Ignoring expected pytest-asyncio teardown issue: {e}")
else:
raise
@pytest.fixture(scope="session")
async def bob_mcp_client(bob_oauth_token: str) -> AsyncGenerator[ClientSession, Any]:
"""MCP client authenticated as bob (viewer role)."""
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp", token=bob_oauth_token, client_name="Bob MCP"
):
yield session
try:
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=bob_oauth_token,
client_name="Bob MCP",
):
yield session
except RuntimeError as e:
if "cancel scope" in str(e) and "different task" in str(e):
logger.debug(f"Ignoring expected pytest-asyncio teardown issue: {e}")
else:
raise
@pytest.fixture(scope="session")
@@ -1208,12 +1236,18 @@ async def charlie_mcp_client(
charlie_oauth_token: str,
) -> AsyncGenerator[ClientSession, Any]:
"""MCP client authenticated as charlie (editor role, in 'editors' group)."""
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=charlie_oauth_token,
client_name="Charlie MCP",
):
yield session
try:
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=charlie_oauth_token,
client_name="Charlie MCP",
):
yield session
except RuntimeError as e:
if "cancel scope" in str(e) and "different task" in str(e):
logger.debug(f"Ignoring expected pytest-asyncio teardown issue: {e}")
else:
raise
@pytest.fixture(scope="session")
@@ -1221,12 +1255,18 @@ async def diana_mcp_client(
diana_oauth_token: str,
) -> AsyncGenerator[ClientSession, Any]:
"""MCP client authenticated as diana (no-access role)."""
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=diana_oauth_token,
client_name="Diana MCP",
):
yield session
try:
async for session in create_mcp_client_session(
url="http://127.0.0.1:8001/mcp",
token=diana_oauth_token,
client_name="Diana MCP",
):
yield session
except RuntimeError as e:
if "cancel scope" in str(e) and "different task" in str(e):
logger.debug(f"Ignoring expected pytest-asyncio teardown issue: {e}")
else:
raise
# Test user/group fixtures for clean test isolation