# 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://localhost: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://localhost: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://localhost: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://localhost:8000/mcp") as (read1, write1, _):
        async with ClientSession(read1, write1) as session1:
            await session1.initialize()

            async with streamablehttp_client("http://localhost: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://localhost: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.