Maintain contextvars between fixtures and tests

pytest_asyncio/plugin.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import contextlib
+import contextvars
 import enum
 import functools
 import inspect
@@ -322,6 +323,23 @@ async def setup():
             res = await gen_obj.__anext__()  # type: ignore[union-attr]
             return res
 
+        context = contextvars.copy_context()
+        setup_task = _create_task_in_context(event_loop, setup(), context)
+        result = event_loop.run_until_complete(setup_task)
+
+        # Copy the context vars set by the setup task back into the ambient
+        # context for the test.
+        context_tokens = []
+        for var in context:
+            try:
+                if var.get() is context.get(var):
+                    # Not modified by the fixture, so leave it as-is.
+                    continue
+            except LookupError:
+                pass
+            token = var.set(context.get(var))
+            context_tokens.append((var, token))
+
         def finalizer() -> None:
             """Yield again, to finalize."""
 
@@ -335,15 +353,40 @@ async def async_finalizer() -> None:
                     msg += "Yield only once."
                     raise ValueError(msg)
 
-            event_loop.run_until_complete(async_finalizer())
+            task = _create_task_in_context(event_loop, async_finalizer(), context)
+            event_loop.run_until_complete(task)
+
+            # Since the fixture is now complete, restore any context variables
+            # it had set back to their original values.
+            while context_tokens:
+                (var, token) = context_tokens.pop()
+                var.reset(token)
 
-        result = event_loop.run_until_complete(setup())
         request.addfinalizer(finalizer)
         return result
 
     fixturedef.func = _asyncgen_fixture_wrapper  # type: ignore[misc]
 
 
+def _create_task_in_context(loop, coro, context):
+    """
+    Return an asyncio task that runs the coro in the specified context,
+    if possible.
+
+    This allows fixture setup and teardown to be run as separate asyncio tasks,
+    while still being able to use context-manager idioms to maintain context
+    variables and make those variables visible to test functions.
+
+    This is only fully supported on Python 3.11 and newer, as it requires
+    the API added for https://github.com/python/cpython/issues/91150.
+    On earlier versions, the returned task will use the default context instead.
+    """
+    try:
+        return loop.create_task(coro, context=context)
+    except TypeError:
+        return loop.create_task(coro)
+
+
 def _wrap_async_fixture(fixturedef: FixtureDef) -> None:
     fixture = fixturedef.func
 
@@ -360,6 +403,10 @@ async def setup():
             res = await func(**_add_kwargs(func, kwargs, event_loop, request))
             return res
 
+        # Since the fixture doesn't have a cleanup phase, if it set any context
+        # variables we don't have a good way to clear them again.
+        # Instead, treat this fixture like an asyncio.Task, which has its own
+        # independent Context that doesn't affect the caller.
         return event_loop.run_until_complete(setup())
 
     fixturedef.func = _async_fixture_wrapper  # type: ignore[misc]

tests/async_fixtures/test_async_fixtures_contextvars.py

@@ -0,0 +1,66 @@
+"""
+Regression test for https://github.com/pytest-dev/pytest-asyncio/issues/127:
+contextvars were not properly maintained among fixtures and tests.
+"""
+
+from __future__ import annotations
+
+import sys
+from contextlib import contextmanager
+from contextvars import ContextVar
+
+import pytest
+
+_context_var = ContextVar("context_var")
+
+
+@contextmanager
+def context_var_manager(value):
+    token = _context_var.set(value)
+    try:
+        yield
+    finally:
+        _context_var.reset(token)
+
+
+@pytest.fixture(scope="function")
+async def no_var_fixture():
+    with pytest.raises(LookupError):
+        _context_var.get()
+    yield
+    with pytest.raises(LookupError):
+        _context_var.get()
+
+
+@pytest.fixture(scope="function")
+async def var_fixture_1(no_var_fixture):
+    with context_var_manager("value1"):
+        yield
+
+
+@pytest.fixture(scope="function")
+async def var_nop_fixture(var_fixture_1):
+    with context_var_manager(_context_var.get()):
+        yield
+
+
+@pytest.fixture(scope="function")
+def var_fixture_2(var_nop_fixture):
+    assert _context_var.get() == "value1"
+    with context_var_manager("value2"):
+        yield
+
+
+@pytest.fixture(scope="function")
+async def var_fixture_3(var_fixture_2):
+    assert _context_var.get() == "value2"
+    with context_var_manager("value3"):
+        yield
+
+
+@pytest.mark.asyncio
+@pytest.mark.xfail(
+    sys.version_info < (3, 11), reason="requires asyncio Task context support"
+)
+async def test(var_fixture_3):
+    assert _context_var.get() == "value3"