Gọi mock async trong python 3.5

Question

Làm thế nào để tôi thực hiện cuộc gọi không đồng bộ từ một coroutine bản địa sang một số khác bằng cách sử dụng unittest.mock.patch?

Tôi hiện đang có một giải pháp vụng về khá:

class CoroutineMock(MagicMock): 
    def __await__(self, *args, **kwargs): 
     future = Future() 
     future.set_result(self) 
     result = yield from future 
     return result 

Sau đó

class TestCoroutines(TestCase): 
    @patch('some.path', new_callable=CoroutineMock) 
    def test(self, mock): 
     some_action() 
     mock.assert_called_with(1,2,3) 

này hoạt động nhưng trông xấu xí. Có cách nào sâu sắc hơn để làm điều này?

Answer 1

Phân lớp MagicMock sẽ tuyên truyền lớp tùy chỉnh của bạn cho tất cả các mocks được tạo ra từ mô hình coroutine của bạn. Ví dụ: AsyncMock().__str__ cũng sẽ trở thành một AsyncMock có lẽ không phải là những gì bạn đang tìm kiếm.

Thay vào đó, bạn có thể muốn xác định nhà máy tạo Mock (hoặc MagicMock) với đối số tùy chỉnh, ví dụ side_effect=coroutine(coro). Ngoài ra, nó có thể là một ý tưởng tốt để tách chức năng coroutine khỏi coroutine (như được giải thích trong documentation).

Dưới đây là những gì tôi đã đưa ra:

from asyncio import coroutine 

def CoroMock(): 
    coro = Mock(name="CoroutineResult") 
    corofunc = Mock(name="CoroutineFunction", side_effect=coroutine(coro)) 
    corofunc.coro = coro 
    return corofunc 

Giải thích về các đối tượng khác nhau:

• corofunc: chức năng coroutine giả
• corofunc.side_effect(): các coroutine, tạo ra cho mỗi cuộc gọi
• corofunc.coro: mô hình được sử dụng bởi coroutine để nhận kết quả
• corofunc.coro.return_value: giá trị trả về bởi các coroutine
• corofunc.coro.side_effect: có thể được sử dụng để nâng cao một ngoại lệ

Ví dụ:

async def coro(a, b): 
    return await sleep(1, result=a+b) 

def some_action(a, b): 
    return get_event_loop().run_until_complete(coro(a, b)) 

@patch('__main__.coro', new_callable=CoroMock) 
def test(corofunc): 
    a, b, c = 1, 2, 3 
    corofunc.coro.return_value = c 
    result = some_action(a, b) 
    corofunc.assert_called_with(a, b) 
    assert result == c 

Answer 2

Một cách khác để chế tạo coroutine là tạo ra coroutine, nó sẽ trả về mô phỏng. Bằng cách này bạn có thể chế tạo các coroutines sẽ được chuyển vào asyncio.wait hoặc asyncio.wait_for.

Điều này làm cho nhiều coroutines phổ mặc dù làm cho thiết lập các bài kiểm tra cồng kềnh hơn:

def make_coroutine(mock) 
    async def coroutine(*args, **kwargs): 
     return mock(*args, **kwargs) 
    return coroutine 


class Test(TestCase): 
    def setUp(self): 
     self.coroutine_mock = Mock() 
     self.patcher = patch('some.coroutine', 
          new=make_coroutine(self.coroutine_mock)) 
     self.patcher.start() 

    def tearDown(self): 
     self.patcher.stop() 

Answer 3

Giải pháp là thực sự khá đơn giản: Tôi chỉ cần thiết để chuyển đổi __call__ phương pháp giả vào coroutine:

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

Điều này hoạt động hoàn hảo, khi giả được gọi, mã nhận được coroutine nguyên bản

Answer 4

Mọi người còn thiếu những gì lẽ là giải pháp đơn giản nhất và rõ ràng nhất:

@patch('some.path') 
def test(self, mock): 
    f = asyncio.Future() 
    f.set_result('whatever result you want') 
    process_smtp_message.return_value = f 
    mock.assert_called_with(1, 2, 3) 

nhớ một coroutine có thể được nghĩ của chỉ là một chức năng được đảm bảo để trả lại một tương lai mà có thể, lần lượt được chờ đợi.