Python 用 Aiohttp 模拟不是模拟

Python Mock with Aiohttp isn't mocking

我正在使用 AioHttp 来实现工作中的服务,在我的测试期间,我正在模拟一个方法,问题是对该方法的调用是调用真实方法而不是模拟方法。

@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

所以我在模拟的断言中得到了这个错误:

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

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

是的,是 class 中的一个可调用方法。没有模拟,测试工作正常。但是我需要模拟对数据库的调用,到目前为止我运气不好。 有什么想法吗?

清单:

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

我是 mocket 的作者,几天前我发布了完全支持 asyncio 的版本 2.0.0 ]/aiohttp.

下面是在 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