Can't async mock a method in a context manager

[MichaelRes]

Code to reproduce the error (with python>3.11)

import asyncio
from unittest.mock import AsyncMock


async def main():
    a = AsyncMock()
    some_object = AsyncMock()
    some_object.__aenter__.return_value = some_object
    some_object.__aexit__.return_value = some_object
    a.some_method.return_value = some_object

    async with a.some_method() as _:
        print("yes")


asyncio.run(main())

error: TypeError: 'coroutine' object does not support the asynchronous context manager protocol

[seifertm]

This is the issue tracker for pytest-asyncio, a pytest plugin to simplify running of asyncio code in tests.

I don't see how your code example has anything to do with pytest-asyncio or event pytest. Maybe you've posted this in the wrong project?

That said, your error message is caused by the fact that a is of type AsyncMock. Change this to a = Mock() and it works.

[diman82]

Get the same error for my test function:

main logic fn:

async def check_website_and_log_error_to_db(website: Website, db_wrapper: PostgresWrapper,
                                            session: aiohttp.ClientSession):
    request_start_time = time.time()
    async with session.get(website.url) as response:
        html = await response.text()
        response_time = time.time()

        if not response.ok:  # == failure, write to db
            website_db = WebsiteDBModel(id=str(uuid.uuid4()), website_url=website.url, website_name=website.name,
                                        request_ts=request_start_time, response_ts=response_time,
                                        status_code=response.status, contents=html,
                                        failure_details='response status not 2XX')
            db_wrapper.write_failed_site(website_db)

test fn:

@pytest.mark.asyncio
async def test_website_check_availability_200():
    website = Website('http://facebook.com', 'Facebook', 10, 'abracadabra2')

    session_mock = MagicMock()
    response_mock = AsyncMock()
    response_mock.status = 200
    session_mock.get = AsyncMock(return_value=response_mock)

    mock_postgres_wrapper = TestingPostgresWrapper()
    res = await check_website_and_log_error_to_db(website, mock_postgres_wrapper, session_mock)

    assert res.status == 200

error:

test_asyncio.py::test_website_check_availability_200 FAILED              [100%]
test_asyncio.py:9 (test_website_check_availability_200)
@pytest.mark.asyncio
    async def test_website_check_availability_200():
        website = Website('http://facebook.com', 'Facebook', 10, 'abracadabra2')
    
        session_mock = MagicMock()
        response_mock = AsyncMock()
        response_mock.status = 200
        session_mock.get = AsyncMock(return_value=response_mock)
    
        mock_postgres_wrapper = TestingPostgresWrapper()
>       res = await check_website_and_log_error_to_db(website, mock_postgres_wrapper, session_mock)

test_asyncio.py:20: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

website = Website(url='http://facebook.com', name='Facebook', check_interval=10, regexp='abracadabra2')
db_wrapper = <tests.tests_postgres.TestingPostgresWrapper object at 0x000001A7B82D5A90>
session = <MagicMock id='1819861148752'>

    async def check_website_and_log_error_to_db(website: Website, db_wrapper: PostgresWrapper,
                                                session: aiohttp.ClientSession):
        request_start_time = time.time()
>       async with session.get(website.url) as response:
E       TypeError: 'coroutine' object does not support the asynchronous context manager protocol

[takarzyna]

Small note to whoever might find it useful:
if you want the mock to work with both async with and await then you need to return AsyncMock in __aenter__, e.g.:

# The main mock object cannot be AsyncMock because then it doesn't work with `async with`
with (patch("module.MyObject") as mock_my_object):
    async def mock_aenter(_):
        # The object returned by `__aenter__` has to be AsyncMock, 
        # and any function you want to mock and call with `await` needs to be defined in it
        mock_my_object_for_await = AsyncMock
        mock_my_object_for_await.call_function = lambda _: return "some value"
        return mock_my_object_for_await

    async def mock_aexit(obj, exc_type, exc, tb):
        return
        
    mock_my_object.__aexit__ = mock_aexit
    mock_my_object.__aenter__ = mock_aenter

    """
    # Then somewhere in what you're testing this will work properly:
    async with MyObject() as my_object:
        return_value = await my_object.call_function("some input value")
    """