Python Mock with Aiohttp isn't mocking

Question

I'm using AioHttp to implement a service at work, and during my tests, I'm mocking a method, the issue is there the call for the method is calling the real instead of the mocked method.

@unittest_run_loop
@patch('export_api.main.add_job_to_db')
async def test_view_job(self, mocked_method):
    json = {
        "edl": "somedata"
    }
    response = await self.client.request("PUT", "/v1/job", json=json)
    mocked_method.assert_called_once_with()
    assert response.status == 200

So I get this error on the assertion of the mock:

msg = "Expected 'add_job_to_db' to be called once. Called 0 times."

My method on the main.py:

 async def __call__(self, request):
    """Faz post do Job na fila do Render"""
    data = await request.json()
    job_id = uuid.uuid4()
    job = Jobs(
        job_id=str(job_id),
        body=data
        )
    try:
        add_job_to_db(self.app['db'], job)
        return web.Response(status=200)
    except DatabaseError as e:
        print(e)
        return web.Response(status=500)

Yes, is a callable method inside a class. The test work fine without the mocking. But I need to mock the call for the db, and I'm not having luck so far. Any ideas?

Answer 1

I'm the author of mocket and few days ago I released the version 2.0.0 which fully supports asyncio/aiohttp.

Here is the same example of code mocking a URL on HTTP and on HTTPS:

import aiohttp
import asyncio
import async_timeout
from unittest import TestCase

from mocket.mocket import mocketize
from mocket.mockhttp import Entry


class AioHttpEntryTestCase(TestCase):
    @mocketize
    def test_http_session(self):
        url = 'http://httpbin.org/ip'
        body = "asd" * 100
        Entry.single_register(Entry.GET, url, body=body, status=404)
        Entry.single_register(Entry.POST, url, body=body*2, status=201)

        async def main(l):
            async with aiohttp.ClientSession(loop=l) as session:
                with async_timeout.timeout(3):
                    async with session.get(url) as get_response:
                        assert get_response.status == 404
                        assert await get_response.text() == body

                with async_timeout.timeout(3):
                    async with session.post(url, data=body * 6) as post_response:
                        assert post_response.status == 201
                        assert await post_response.text() == body * 2

        loop = asyncio.get_event_loop()
        loop.set_debug(True)
        loop.run_until_complete(main(loop))

    @mocketize
    def test_https_session(self):
        url = 'https://httpbin.org/ip'
        body = "asd" * 100
        Entry.single_register(Entry.GET, url, body=body, status=404)
        Entry.single_register(Entry.POST, url, body=body*2, status=201)

        async def main(l):
            async with aiohttp.ClientSession(loop=l) as session:
                with async_timeout.timeout(3):
                    async with session.get(url) as get_response:
                        assert get_response.status == 404
                        assert await get_response.text() == body

                with async_timeout.timeout(3):
                    async with session.post(url, data=body * 6) as post_response:
                        assert post_response.status == 201
                        assert await post_response.text() == body * 2

        loop = asyncio.get_event_loop()
        loop.set_debug(True)
loop.run_until_complete(main(loop))

Source: https://github.com/mindflayer/python-mocket/blob/master/tests/tests35/test_http_aiohttp.py

Answer 2

Checklist:

1. Use asynctest package
2. Don't forget to use @asyncio.coroutine decorator around the unit test
3. Patch the object where it is used, in this case it export_api.main.add_job_to_db