带有Aiohttp的Python Mock不是mocking

Question

我在工作中使用AioHttp来实现一个服务，在我的测试中，我模拟了一个方法，问题是对该方法的调用是调用real而不是mocked方法。在

@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

所以我在mock的断言中得到了这个错误：

^{pr2}$

我的方法主.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)

是的，是类中的可调用方法。这个测试没有嘲笑的效果很好。但我需要嘲笑db的要求，我到目前为止运气不好。 有什么想法吗？在

Answer 1

检查清单：

1. 使用asynctest包
2. 不要忘记在单元测试中使用@asyncio.coroutine装饰器
3. 修补使用它的对象，在本例中它export_api.main.add_job_to_db

Answer 2

我是mocket的作者，几天前我发布了完全支持异步/aiohttp的版本2.0.0。在

下面是在HTTP和HTTPS上模拟URL的相同示例：

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))

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