-
Notifications
You must be signed in to change notification settings - Fork 41
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
MagicMock doesn't implement __aenter__
or __aexit__
#29
Comments
Hi, Thanks for the report, it's on the todo list! |
+1 it would be really useful to add implementation of |
I add this link as relevant to the conversation http://bugs.python.org/issue26467 |
Hi, guys! Any news about this issue? |
I would like to mock aiopg.sa.create_engine with CoroutineMock, but if I use async-with-style it does not work import asyncio
import aiopg.sa
import asynctest
mock = asynctest.CoroutineMock()
async def create_engine():
kwargs = {'echo': True}
return await aiopg.sa.create_engine(
'dbname=db user=user password=pass host=127.0.0.1',
**kwargs
)
async def go(get_engine):
engine = await get_engine()
async with engine.acquire() as conn:
async with conn.begin():
r = await conn.execute('SELECT version();')
return await r.fetchone()
row = asyncio.get_event_loop().run_until_complete(go(create_engine))
print(row)
row = asyncio.get_event_loop().run_until_complete(go(mock))
print(row)
but if I do not use async-with-style all works fine import asyncio
import aiopg.sa
import asynctest
mock = asynctest.CoroutineMock()
async def create_engine():
kwargs = {'echo': True}
return await aiopg.sa.create_engine(
'dbname=wallet user=wallet password=qazwsx host=127.0.0.1',
**kwargs
)
async def go(get_engine):
engine = await get_engine()
conn = await engine.acquire()
trans = await conn.begin()
r = await conn.execute('SELECT version();')
await trans.commit()
engine.release(conn)
return await r.fetchone()
row = asyncio.get_event_loop().run_until_complete(go(create_engine))
print(row)
row = asyncio.get_event_loop().run_until_complete(go(mock))
print(row)
|
We had the same problem mocking aiopg. We ended up with this workaround: class AsyncContextManagerMock(MagicMock):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
for key in ('aenter_return', 'aexit_return'):
setattr(self, key, kwargs[key] if key in kwargs else MagicMock())
async def __aenter__(self):
return self.aenter_return
async def __aexit__(self, *args):
return self.aexit_return |
Hello, Some news regarding this issue: I'm working on a solution which seems to be working for the basic needs (in the async_with branch). Some cases are yet to be tested, but if some of you want to try it, feedback is welcome. |
Hello, @Martiusweb ! Do you mean async_magic branch ? |
Yes, I mean async_magic (https://github.com/Martiusweb/asynctest/tree/async_magic). |
I have just try async_magic branch with code as below import asyncio
import aiopg.sa
import asynctest
mock = asynctest.CoroutineMock()
async def create_engine():
kwargs = {'echo': True}
return await aiopg.sa.create_engine(
'dbname=db user=dbuser password=qazwsx host=127.0.0.1',
**kwargs
)
async def go(get_engine):
engine = await get_engine()
async with engine.acquire() as conn:
async with conn.begin():
r = await conn.execute('SELECT version();')
return await r.fetchone()
row = asyncio.get_event_loop().run_until_complete(go(create_engine))
print(row)
row = asyncio.get_event_loop().run_until_complete(go(mock))
print(row) I have got the following: python atest.py |
This is because asynctest has no way to know that engine.acquire() is a coroutine, hence it's mocked as a MagicMock rather than a CoroutineMock. However, it worked before because CoroutineMock returned an other CoroutineMock as the result of its invocation, which is wrong. You can fix this by doing something like:
|
import asyncio
import aiopg.sa
import asynctest
mock = asynctest.CoroutineMock(return_value=asynctest.MagicMock())
async def create_engine():
kwargs = {'echo': True}
return await aiopg.sa.create_engine(
'dbname=db user=dbuser password=qazwsx host=127.0.0.1',
**kwargs
)
async def go(get_engine):
engine = await get_engine()
async with engine.acquire() as conn:
async with conn.begin():
r = await conn.execute('SELECT version();')
return await r.fetchone()
row = asyncio.get_event_loop().run_until_complete(go(create_engine))
print(row)
row = asyncio.get_event_loop().run_until_complete(go(mock))
print(row) python atest.py |
That's still the same issue: asynctest can't know that the This:
is already equivalent to the default behavior, but you can do this:
(edited: first sentence was incomplete) |
Ok. I just modify code like this import asyncio
import aiopg.sa
import asynctest
def make_engine(connection):
return asynctest.CoroutineMock(
return_value=asynctest.CoroutineMock(
acquire=asynctest.CoroutineMock(
return_value=connection
)
)
)
async def create_engine():
kwargs = {'echo': True}
return await aiopg.sa.create_engine(
'dbname=dbname user=dbuser password=dbpassword host=127.0.0.1',
**kwargs
)
async def go(get_engine):
engine = await get_engine()
print(engine.acquire)
print(asyncio.iscoroutinefunction(engine.acquire))
print(asyncio.iscoroutine(engine.acquire()))
async with engine.acquire() as conn:
async with conn.begin():
r = await conn.execute('SELECT version();')
return await r.fetchone()
#row = asyncio.get_event_loop().run_until_complete(go(create_engine))
#print(row)
row = asyncio.get_event_loop().run_until_complete(go(make_engine(asynctest.CoroutineMock())))
print(row) so, mock have an acquire attribute which is a Coroutine. but any way I have got the following:
|
You must use MagicMock instead of CoroutineMock: Engine, _EngineAcquireContextManager, and _ConnectionContextManager are not coroutines. Anyway, MagicMock is not smart enough to do what you want to achieve: you the mock system to infer the type of the result of engine.acquire() and conn.begin(), and it's not possible in pure Python. It would be the same problem with a non-asynchronous equivalent. If you want to mock all the chain, you must do something like:
|
I implemented aenter, aexit, aiter and anext. I think there are cases left to test, especially cases dealing with exceptions (eg: when an exception is raised inside an asynchronous context manager, the exception seems to be swallowed and lost). It's also tedious to mock those magic methods, because the async context manager class must be used as mock spec. Classes like |
@Martiusweb , thanks a lot! The following code works fine import asyncio
import aiopg.sa
import asynctest
def make_engine():
connection_mock = asynctest.MagicMock(
execute=asynctest.CoroutineMock(
return_value=asynctest.CoroutineMock(
fetchone=asynctest.CoroutineMock()
)
)
)
connection_mock.begin.return_value = asynctest.MagicMock(
aiopg.sa.connection._SAConnectionContextManager)
acquire_mock = asynctest.MagicMock(
aiopg.sa.engine._EngineAcquireContextManager)
acquire_mock.__aenter__.return_value = connection_mock
engine_mock = asynctest.MagicMock(aiopg.sa.engine.Engine)
engine_mock.acquire.return_value = acquire_mock
return asynctest.CoroutineMock(return_value=engine_mock)
async def create_engine():
kwargs = {'echo': True}
return await aiopg.sa.create_engine(
'dbname=dbname user=dbuser password=dbpasssword host=127.0.0.1',
**kwargs
)
async def go(get_engine):
engine = await get_engine()
async with engine.acquire() as conn:
async with conn.begin():
r = await conn.execute('SELECT version();')
return await r.fetchone()
row = asyncio.get_event_loop().run_until_complete(go(create_engine))
print(row)
row = asyncio.get_event_loop().run_until_complete(go(make_engine()))
print(row) the result is
but on branch async_magic I have seen message about Unclosed connection when execute row = asyncio.get_event_loop().run_until_complete(go(create_engine)) Note, on current version from pypi I have not seen this message. |
Hi, This is likely because somewhere in your code an actual connection is created but never closed. You'll have to track which part of your test uses real code rather than the mock. |
No no no!!! the following code does not use MOCK at all import asyncio
import aiopg.sa
import asynctest
'''
def make_engine():
connection_mock = asynctest.MagicMock(
execute=asynctest.CoroutineMock(
return_value=asynctest.CoroutineMock(
fetchone=asynctest.CoroutineMock()
)
)
)
connection_mock.begin.return_value = asynctest.MagicMock(
aiopg.sa.connection._SAConnectionContextManager)
acquire_mock = asynctest.MagicMock(
aiopg.sa.engine._EngineAcquireContextManager)
acquire_mock.__aenter__.return_value = connection_mock
engine_mock = asynctest.MagicMock(aiopg.sa.engine.Engine)
engine_mock.acquire.return_value = acquire_mock
return asynctest.CoroutineMock(return_value=engine_mock)
'''
async def create_engine():
kwargs = {'echo': True}
return await aiopg.sa.create_engine(
'dbname=db user=user password=passwd host=127.0.0.1',
**kwargs
)
async def go(get_engine):
engine = await get_engine()
async with engine.acquire() as conn:
async with conn.begin():
r = await conn.execute('SELECT version();')
return await r.fetchone()
row = asyncio.get_event_loop().run_until_complete(go(create_engine))
print(row)
#row = asyncio.get_event_loop().run_until_complete(go(make_engine()))
#print(row) and when I execute this code on current version from pypi everything is OK, but when I execute this code on the async_magic branch I have seen
|
U can test it yourself. Just switch to current version from pypi and execute the code above. Then switch to async_magic branch and try to execute this code. U will see message about Unclosed connection |
In your example, you don't use asynctest at all. I don't see what I can do for you there. Since it's not related to the async context managers mocking issue, if you think it's related to asynctest and can provide more details (for instance, use git bisect to find the commit which introduced a bug), please open a new issue. |
I just want to say, that on current version from pypi and on async_magic result of execution the above code is different. The only difference is asynctest version |
Executing the above code in both branches master and async_magic leads to message about Unclosed connection. But if I use current version from pypi https://pypi.python.org/pypi/asynctest/0.10.0 everything is ok |
If I switch to commit a3b7b5e on master (the current version in pypi) everything is OK. If I switch back to master I have got message about Unclosed connection |
Wow! If I switch back to one commit from current master
everything is OK executing the code above. No message about Unclosed connection. If I switch to the HEAD again , I have got message about Unclosed connection. So, something wrong with commit e005798 This is about commit which introduced a bug (strange behavior). |
For reference: If I comment importing asynctest in the code above everything is OK on master (no message about Unclosed connection). import asyncio
import aiopg.sa
# import asynctest
... |
@Martiusweb what are the plans for this issue? Are |
Yes, I've been working on this but there are still some parts I need to fix before I can merge the PR: You can try the branch though, as the most common use cases are working. |
nice work @Martiusweb on a real fix and @Galbar on work-around, this thread inspired us to do a simpler fix: class MagicMockContext(MagicMock):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
type(self).__aenter__ = CoroutineMock(return_value=MagicMock())
type(self).__aexit__ = CoroutineMock(return_value=MagicMock()) with usage: with asynctest.mock.patch('dummy.Dummy', new_callable=MagicMockContext) as mockClientSession: |
@Martiusweb any chance the branch could be merged? I've been using it for a while and works for me. If not, what are the missing things? Maybe other people can help there. |
I'll review my work and list what's missing by the end of the week. Exception raised in the context manager are not propagated correctly yet: I need to write the tests to ensure it works. I also would like to add an API which helps manipulating those mocks, but I'll probably do that in an other PR. |
Cool, once the list is available I'll check if there is something I can help this. I prefer installing things from a released version rather than from specific branch/commit :P |
This issue has been fixed in the 0.11.0 release which has just been pushed to pypi. Feedback is more than welcome! If you see anything buggy or that can be improved, open a new bug! |
Good news, thanks a lot! :) |
With asynctest==0.11.0 this should work, right?:
However I get: Am I doing something wrong? For completeness:
|
You must think of CoroutineMock as the mock of a coroutine function, nothing else. In your case, you should use MagicMock, not CoroutineMock. (Note to self: I really need to work on the guide) |
Sorry for being dense, but this:
gives this: Can you please help me? |
You spotted a bug: unless you specify a spec implementing |
@x0zzz the bug is fixed in v0.11.1 |
Thank you so very much :) |
btw just a q in case anyone knows off the top of their head, is there any quick way to ensure that |
Maybe I misunderstood your comment, but this is not the default behavior of unittest:
The simplest way to do this is to set the return_value:
|
When testing asyncio code, the probability of needing to test async context manager is high.
The current code using MagicMock doesn't implement
__aenter__
or__aexit__
methodsThe text was updated successfully, but these errors were encountered: