ContextVar not propagated from fixture to test

[beedub]

I'm trying to figure out why this code sample isn't working. I'm not experienced with asyncio yet, so this is very likely a simple misunderstanding on my part.

from _contextvars import ContextVar

import pytest

blah = ContextVar("blah")


@pytest.fixture
async def my_context_var():
    blah.set("hello")
    assert blah.get() == "hello"
    yield blah


@pytest.mark.asyncio
async def test_blah(my_context_var):
    assert my_context_var.get() == "hello" # this fails

[kalefranz]

I've also hit this. I think what's happening is explained by

pytest-asyncio/pytest_asyncio/plugin.py

Line 97 in 2c8023d

return loop.run_until_complete(setup())

along with understanding of https://www.python.org/dev/peps/pep-0550. The setup() function there executes the code within the coroutine fixture. Using loop.run_until_complete(setup()) means that context changes won't "leak" outside of the execution context within the coroutine.

One option is to try to avoid using a coroutine fixture when you need to set context. So for example

import asyncio
from contextvars import ContextVar
import pytest

blah = ContextVar("blah")


async def set_context():
    blah.set("hello")
    assert blah.get() == "hello"


@pytest.fixture
def my_context_var():
    asyncio.run(set_context())
    assert blah.get() == "hello"
    yield blah


@pytest.mark.asyncio
async def test_blah(my_context_var):
    assert my_context_var.get() == "hello"

[Andrei-Pozolotin]

+1

[nolar]

A note on the workaround above: the set_context() coroutine will run in its own context, and its contextvar will not get into the surrounding/sibling coroutines. The example above fails with:

__________________________________________________________________________ ERROR at setup of test_blah ___________________________________________________________________________

    @pytest.fixture
    def my_context_var():
        asyncio.run(set_context())
>       assert blah.get() == "hello"
E       LookupError: <ContextVar name='blah' at 0x102ebb900>

A much easier solution is to set the contextvar directly in the sync fixture — note, not async fixture, but sync:

from contextvars import ContextVar
import pytest

blah = ContextVar("blah")


@pytest.fixture
def my_context_var():  # << SYNC!!!
    blah.set("hello")
    assert blah.get() == "hello"
    yield blah


@pytest.mark.asyncio
async def test_blah(my_context_var):
    assert my_context_var.get() == "hello"

I use this approach in Kopf actively, and it works:

• https://github.com/zalando-incubator/kopf/blob/0.26/tests/posting/conftest.py
• https://github.com/zalando-incubator/kopf/blob/0.26/tests/conftest.py#L322-L344
• https://github.com/zalando-incubator/kopf/blob/0.26/tests/authentication/test_credentials.py#L91-L95
• And basically everything you can find there with _var.set(

[dimaqq]

The cause appears to be that the async fixtures are ran in own Tasks.

To augment @nolar 's solution, I'd recommend this:

@pytest.fixture
def fake_db(database):
    token = DB.set(database)
    yield
    DB.reset(token)

@pytest.fixture
async def database(...):
    ...

[butla]

I have async functions that set context vars with objects that should be used in the entire app (I use a file with context vars as an IoC container). I wanted to run that code in a fixture, so I get nice setup and teardown, but sadly, I can't run that in a sync fixture. So the workarounds don't work for me.

Just letting anybody interested know :)

[nikicat]

This issue could be fixed easily in Python 3.11, but even for 3.10 this could be fixed like this