async method fixture is run with different self than the test method

[Incanus3]

Hi,
I just ran into this - if you make a fixture method async, it is run with a different self, so you can't set attributes on self and read them in test methods. Here is a simple snippet to reproduce the problem:

import pytest

pytestmark = pytest.mark.asyncio

class TestAsyncFixture:
  @pytest.fixture #(autouse = True)
  async def setup(self):
    # await self.client._handle_ws_request(self.payment_start_request)
    self.some_attr = 'value'

    print()
    print('in fixture')
    print(self)
    print(self.__dict__)

  async def test_response(self, setup):
    print('in test')
    print(self)          # different object
    print(self.__dict__) # some_attr missing

At first, I thought this only concerns autouse fixtures, but now I tried using it without autouse and it has the same problem. Is this a known issue, or is it supposed to work?

I'm using

python            3.8.3
pytest            6.1.2
pytest-asyncio    0.14.0

Btw, thanks for this great library, I'm using it heavily and it's been of great help.

[Tinche]

I was not aware of this. Could you look into it and submit a PR? Thanks.

[Incanus3]

I'm afraid that's beyond my abilities. I went over the code, but 1) I don't understand a lot of it, since I'm completely unfamiliar with pytest internals, hooks, etc. and 2) none of it seems related to me - AFAIU none of your code concerns class-based tests, instantiation of the test classes and passing the instance btw fixture methods and test methods.

[seifertm]

@Incanus3 I understand that the plugin code might be hard to understand at first. But on second glance, it's not that hard :)
Let me provide a few pointers.

The discovery of test cases is performed by pytest, not pytest-asyncio. Pytest's responsibilities include finding fixtures and test cases in class-based tests. That's why you cannot find anything concerning class-based tests in pytest-asyncio. Pytest's plugin system allows us to take action before or after a fixture or test case is executed.

For example, the function pytest_fixture_setup is used to modify the fixture initialization:

pytest-asyncio/pytest_asyncio/plugin.py

Lines 81 to 143 in 1bb7f30

	@pytest.hookimpl(hookwrapper=True)
	def pytest_fixture_setup(fixturedef, request):
	"""Adjust the event loop policy when an event loop is produced."""
	if fixturedef.argname == "event_loop":
	outcome = yield
	loop = outcome.get_result()
	policy = asyncio.get_event_loop_policy()
	policy.set_event_loop(loop)
	return
	if isasyncgenfunction(fixturedef.func):
	# This is an async generator function. Wrap it accordingly.
	generator = fixturedef.func
	fixture_stripper = FixtureStripper(fixturedef)
	fixture_stripper.add(FixtureStripper.EVENT_LOOP)
	fixture_stripper.add(FixtureStripper.REQUEST)
	def wrapper(*args, **kwargs):
	loop = fixture_stripper.get_and_strip_from(FixtureStripper.EVENT_LOOP, kwargs)
	request = fixture_stripper.get_and_strip_from(FixtureStripper.REQUEST, kwargs)
	gen_obj = generator(*args, **kwargs)
	async def setup():
	res = await gen_obj.__anext__()
	return res
	def finalizer():
	"""Yield again, to finalize."""
	async def async_finalizer():
	try:
	await gen_obj.__anext__()
	except StopAsyncIteration:
	pass
	else:
	msg = "Async generator fixture didn't stop."
	msg += "Yield only once."
	raise ValueError(msg)
	loop.run_until_complete(async_finalizer())
	request.addfinalizer(finalizer)
	return loop.run_until_complete(setup())
	fixturedef.func = wrapper
	elif inspect.iscoroutinefunction(fixturedef.func):
	coro = fixturedef.func
	fixture_stripper = FixtureStripper(fixturedef)
	fixture_stripper.add(FixtureStripper.EVENT_LOOP)
	def wrapper(*args, **kwargs):
	loop = fixture_stripper.get_and_strip_from(FixtureStripper.EVENT_LOOP, kwargs)
	async def setup():
	res = await coro(*args, **kwargs)
	return res
	return loop.run_until_complete(setup())
	fixturedef.func = wrapper
	yield

It is called for each fixture and receives a fixturedef argument. fixturedefis a pytest type that describes a fixture and contains the fixture function object, among other information.

Now let's have a look at pytest_fixture_setup in detail: First, there is a special treatment for the event_loop:

pytest-asyncio/pytest_asyncio/plugin.py

Line 84 in 1bb7f30

if fixturedef.argname == "event_loop":

Then it checks if the fixture is any async generator:

pytest-asyncio/pytest_asyncio/plugin.py

Line 91 in 1bb7f30

if isasyncgenfunction(fixturedef.func):

)
However, this is not the case for your setup function, because it does not yield any value.

The last check is whether your fixture is a coroutine function.

pytest-asyncio/pytest_asyncio/plugin.py

Lines 127 to 143 in 1bb7f30

	elif inspect.iscoroutinefunction(fixturedef.func):
	coro = fixturedef.func
	fixture_stripper = FixtureStripper(fixturedef)
	fixture_stripper.add(FixtureStripper.EVENT_LOOP)
	def wrapper(*args, **kwargs):
	loop = fixture_stripper.get_and_strip_from(FixtureStripper.EVENT_LOOP, kwargs)
	async def setup():
	res = await coro(*args, **kwargs)
	return res
	return loop.run_until_complete(setup())
	fixturedef.func = wrapper
	yield

This check should be true, so the function replaces the fixture with a wrapper. The wrapper retrieves an event loop, awaits the fixture, and returns the result.

The problem seems to be that the wrapper references a different self. It might be worth looking into functools.wrap and see if it solves the issue.

Did that make things clearer?

[Incanus3]

Hi and thanks @seifertm, this is pretty helpful, even though I understood most of this before, at least conceptually (I didn't know how exactly does pytest.hookimpl work, but from your pointers I probably don't need to).

When I looked at this before, I actually made a mistake that's not directly related to pytest, but rather stems from my apparently not-so-great knowledge of asyncio - I thought the inspect.iscoroutinefunction branch would only be executed for the old-style @asyncio.coroutine decorated functions, not async ones, so I though if the name of the arg is not event_loop, it's not a generator function and it's not an asyncio.coroutine (which I now found out all async functions are) none of the branches takes effect.

The second thing I was struggling with still holds though - even though I can intercept the fixture initialization and replace it with a wrapper (or potentially a different function altogether), I don't see how I can influence the way the function is actually called later in the process. I went back to the code, added a small test and added a few debug prints to the pytest_fixture_setup function and lo and behold, the coroutine function branch indeed gets called, but if I print out the coro (i.e. fixturedef.func), I can see it is already a bound to the "wrong" object, that is, a different object than the self that is passed to the test method. I could do with this function anything I wanted, but since I don't have access to the "right" test class instance, I have no way of rebinding the function.

[seifertm]

I can reproduce that that the self object is the same in the setup and pytest_fixture_setup, but different in the test case.

However, I currently have no idea why this should be the case. It seems not to be happening when using regular synchronous methods.

[asvetlov]

@seifertm can I kindly ask you to check the self instance behavior for non-asyncio test cases?
Is it the same or not? Does pytest-asyncio break the standard rule?

[seifertm]

Good question. I checked once with async test functions and once with non-async tests.

It seems that pytest-asyncio behaves differently here.

The output of the async tests:

in fixture
<test_self.TestAsyncFixture object at 0x7f77cf9727f0>
{'some_attr': 'value'}
in test
<test_self.TestAsyncFixture object at 0x7f77cf55a670>
{}

Output for the non-async tests:

in fixture
<test_self.TestAsyncFixture object at 0x7f0129c76670>
{'some_attr': 'value'}
in test
<test_self.TestAsyncFixture object at 0x7f0129c76670>
{'some_attr': 'value'}

[mjpieters]

I came here to report this same bug.

Pytest resolves fixtures that are bound methods each time a test method is run, by using the _pytest.fixtures.resolve_fixture_function() utility function, which looks for request.instance (where request is a FixtureRequest instance). If that attribute is set, it'll check if __self__ exists on the fixture and if the type of __self__ is compatible with the instance (isinstance(request.instance, fixturefunc.__self__.__class__)) before using fixturefunc.__get__(request.instance). When the fixture definition flag unittest is set, the isinstance() check is skipped, the method is rebound unconditionally in that case.

Now, this re-wrapping fails for pytest-asyncio fixtures because the _wrap_async wrapper doesn't have a __self__ attribute, nor can it be used as a bound method anyway. That's to be expected.

All this means there are two ways this can be fixed:

1. If _wrap_async is passed a bound fixture method, have it unwrap the fixture function (use original.__func__) and if you had to unwrap, you need to rebind the _async_fixture_wrapper to the original __self__ attribute (use .__get__(original.__self__)) and have the wrapper pass on arbitrary positional arguments (*args) to the unbound fixture function. This way pytest's fixture re-binding using resolve_fixture_function() will work again as designed.
2. Have _wrap_async resolve the fixture binding in the _async_fixture_wrapper; it has access to the same request, and so can rebind func inside _async_fixture_wrapper. It can't quite reuse resolve_fixture_function() here as you already replaced the bound fixture method, so you'd either have to re-implement that function or create a synthetic FixtureDef with the original bound fixture method.

Here is a version that re-implements the utility function, inline, each time it is called; I dropped the isinstance check the original makes, for simplicity:

def _wrap_async(func) -> None:
    @functools.wraps(func, assigned=(*functools.WRAPPER_ASSIGNMENTS, "__self__"))
    def _async_fixture_wrapper(
        event_loop: asyncio.AbstractEventLoop, request: SubRequest, **kwargs: Any
    ) -> _R:
        fixturefunc = func
        if request.instance is not None:
            # rebind bound fixture method to active instance
            try:
                unbound, self = fixturefunc.__func__, fixturefunc.__self__
            except AttributeError:
                pass
            else:
                fixturefunc = unbound.__get__(request.instance)

        async def setup() -> _R:
            res = await fixturefunc(**_add_kwargs(fixturefunc, kwargs, event_loop, request))
            return res

        return event_loop.run_until_complete(setup())

    return _async_fixture_wrapper

Using that version of _wrap_async fixes this bug for me.

I'll come up with a more complete implementation in a PR.

[seifertm]

Thank you for the great explanation of the root cause!