"Attempted to exit cancel scope in a different task than it was entered in" when used with pytest_asyncio.fixture decorator

[aayushchhabra1999]

Describe the bug

Using aconnect_ws with @pytest_asyncio.fixture triggers
RuntimeError: Attempted to exit cancel scope in a different task than it was entered in

To Reproduce

NOTE: I am using the patch described in this issue: #72

@pytest_asyncio.fixture
async def websocket_connection(
    kniru_api_async_client: AsyncClient,
    bearer_auth_header: dict,
) -> AsyncGenerator[AsyncWebSocketSession, None]:
    async with aconnect_ws(
        f"url",
        kniru_api_async_client,
    ) as kniru_chat_ws:
        yield kniru_chat_ws
        
@pytest.mark.asyncio
@pytest.mark.usefixtures("create_current_user")
async def test_chat_websocket_connection(websocket_connection: AsyncWebSocketSession):
    """Test that the connection can be established and closed."""
    await websocket_connection.close()

Expected behavior

Similar behavior to manually using this part in the tests directly. Works perfectly without declaring as fixture.

Configuration

• Python version: 3.11
• httpx-ws version: 0.6

[frankie567]

It looks like a misuse of pytest-asyncio, which creates dedicated event loop by default since 0.21 instead of a single global one: the event loop in your fixture is not the same as the one in the test, hence the error you get.

Check pytest-asyncio doc to see how to work around that: https://pytest-asyncio.readthedocs.io/en/latest/how-to-guides/change_fixture_loop.html

[jeremy656]

I faced with the same issue, but from your answer, I don't understand what needs to be done.

below listed my simple app

main.py

@asynccontextmanager
async def lifespan(fast_app: FastAPI):
    client = AsyncIOMotorClient("mongodb://localhost:27018")
    await init_beanie(database=client.db_name, document_models=[Product])
    yield


app = FastAPI(
    lifespan=lifespan
)


class Category(BaseModel):
    name: str
    description: str


class Product(Document):
    name: str
    description: Optional[str] = None
    price: Indexed(float)
    category: Category


@app.websocket("/hello_ws")
async def websocket_endpoint(websocket: WebSocket, ):
    await websocket.accept()
    await websocket.send_json({'hello': 'world'})

    while True:
        await websocket.receive_json()
        data = await Product.find().to_list()
        await websocket.send_json(data)

conftest.py

@pytest_asyncio.fixture
async def api_client_for_ws(fast_api: FastAPI):
    """api client fixture."""
    server_name = "https://localhost"
    async with AsyncClient(transport=ASGIWebSocketTransport(app=fast_api), base_url=server_name) as ac:
        yield ac


@pytest_asyncio.fixture
async def websocket_client(api_client_for_ws: AsyncClient, fast_api: FastAPI):
    base_ws_endpoint = fast_api.url_path_for('websocket_endpoint')
    async with aconnect_ws(base_ws_endpoint, api_client_for_ws) as websocket:
        yield websocket


@pytest_asyncio.fixture
async def api_client(fast_api: FastAPI):
    """api client fixture."""
    server_name = "https://localhost"
    async with AsyncClient(transport=ASGITransport(app=fast_api), base_url=server_name) as ac:
        yield ac


@pytest_asyncio.fixture
async def fast_api():
    from main import app
    async with LifespanManager(app, startup_timeout=100, shutdown_timeout=100):
        yield app

test.py

@pytest.mark.asyncio
async def test_websocket(websocket_client):
    data = await websocket_client.receive_json()
    assert data == {'hello': 'world'}

    send_data = {"type": "product.list"}
    await websocket_client.send_json(send_data)
    received_data = await websocket_client.receive_json()
    assert received_data == []
    await websocket_client.close()

Configuration

Python version: 3.10

Requirements

asgi-lifespan==2.1.0
beanie==1.27.0
fastapi==0.115.0
httpx-ws==0.6.2
httpx==0.27.2
pytest==8.3.3
pytest-asyncio==0.24.0
pytest_mock==3.14.0

[aayushchhabra1999]

Hey @jeremy656
This is what I did to make things work. At the root of your project, update your conftest.py to include the following hook:

# conftest.py
import pytest
from pytest_asyncio import is_async_test

def pytest_collection_modifyitems(items) -> None:  # noqa: ANN001
    """Set the loop scope to session for async tests."""
    pytest_asyncio_tests = (item for item in items if is_async_test(item))
    session_scope_marker = pytest.mark.asyncio(loop_scope="session")
    for async_test in pytest_asyncio_tests:
        async_test.add_marker(session_scope_marker, append=False)

You can read more at pytest-dev/pytest-asyncio#61

If this was your issue as well, it might be helpful to include a recipe/cookbook section in the docs for folks. (CC: @frankie567 )

[ncoghlan]

@frankie567 I'm able to reproduce this with just an async generator and asyncio.run (which closes any open async generators when it exits):

>>> from httpx_ws import aconnect_ws
>>> import asyncio
>>> WS_URL = "ws://localhost:1234/system"
>>> async def open_ws():
...     async with aconnect_ws(WS_URL):
...         yield
...
>>> async def leave_gen_running():
...     async for _ in open_ws():
...         # Let the loop shutdown handle the gen cleanup
...         break
...
>>> asyncio.run(leave_gen_running())
unhandled exception during asyncio.run() shutdown
task: <Task finished name='Task-16' coro=<<async_generator_athrow without __name__>()> exception=RuntimeError('Attempted to exit cancel scope in a different task than it was entered in')>
Traceback (most recent call last):
  File "<stdin>", line 2, in open_ws
  File "/usr/lib64/python3.12/contextlib.py", line 231, in __aexit__
    await self.gen.athrow(value)
  File "/home/acoghlan/project/.venv/lib64/python3.12/site-packages/httpx_ws/_api.py", line 1320, in aconnect_ws
    async with _aconnect_ws(
               ^^^^^^^^^^^^^
  File "/usr/lib64/python3.12/contextlib.py", line 231, in __aexit__
    await self.gen.athrow(value)
  File "/home/acoghlan/project/.venv/lib64/python3.12/site-packages/httpx_ws/_api.py", line 1238, in _aconnect_ws
    async with session:
               ^^^^^^^
  File "/home/acoghlan/project/.venv/lib64/python3.12/site-packages/httpx_ws/_api.py", line 648, in __aexit__
    await self._exit_stack.aclose()
  File "/usr/lib64/python3.12/contextlib.py", line 696, in aclose
    await self.__aexit__(None, None, None)
  File "/usr/lib64/python3.12/contextlib.py", line 754, in __aexit__
    raise exc_details[1]
  File "/usr/lib64/python3.12/contextlib.py", line 737, in __aexit__
    cb_suppress = await cb(*exc_details)
                  ^^^^^^^^^^^^^^^^^^^^^^
  File "/home/acoghlan/project/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 778, in __aexit__
    return self.cancel_scope.__exit__(exc_type, exc_val, exc_tb)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/acoghlan/project/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 456, in __exit__
    raise RuntimeError(
RuntimeError: Attempted to exit cancel scope in a different task than it was entered in

(I originally encountered the exception when writing some doctests for an SDK, but initially thought it might be a bad interaction with the Sphinx doctest machinery. This minimal reproducer came from trying to eliminate the doctest failure by using a generator of my own to force the cleanup, but it didn't resolve the issue (since the aclose still ends up coming from a different place than where the websocket was originally created)

[ncoghlan]

As far as I can tell, the restriction is being introduced by combining the background ping and keepalive tasks into an anyio task group, but it isn't clear what is being gained from that vs cancelling the two tasks individually (which should be entirely compatible with loop.shutdown_asyncgens).

[ncoghlan]

That said, https://peps.python.org/pep-0789/#in-a-user-defined-context-manager does cite the websocket background task scenario as one of the motivating examples for being able to prevent yields inside a managed context.