Python Utility for asynchronous mocking

Python Utility for asynchronous mocking.py

import asyncio
from unittest import mock


class AsyncMock(mock.MagicMock):
    @asyncio.coroutine
    def __call__(self, *args, **kwargs):
        return super(AsyncMock, self).__call__(*args, **kwargs)

# Alternative using Tornado...

class AsyncMock(mock.MagicMock):
    async def __call__(self, *args, **kwargs):
        return super().__call__(*args, **kwargs)

def empty_http_request(url="localhost", headers={}):
    return HTTPRequest(url=url, headers=HTTPHeaders(**headers))


def empty_http_response(body="", code=200, error=None):
    return HTTPResponse(
        request=empty_http_request(),
        headers=HTTPHeaders(),
        code=code,
        buffer=BytesIO(body.encode()),
        error=error
    )


def response_to_future(response):
    f = Future()
    f.set_result(response)
    return f


class StaleClientTests(testing.AsyncTestCase):

    def setUp(self):
        super(StaleClientTests, self).setUp()

    @mock.patch("bf_stale_client.client.api_gateway")
    @testing.gen_test
    async def test_api_gateway_call(self, mock_api_gateway):
        response = empty_http_response(body="body")
        request = empty_http_request(headers={"x-auth-token": "foo"})

        http_client = mock.MagicMock()
        http_client.fetch.return_value = response_to_future(empty_http_response("body"))

        # TODO: change arg to response
        mock_api_gateway.get = AsyncMock(return_value=empty_http_response("body"))