Làm thế nào để giả lập asyncio coroutines?

Question

Mã sau không thành công với TypeError: 'Mock' object is not iterable trong ImBeingTested.i_call_other_coroutines vì tôi đã thay thế ImGoingToBeMocked bằng đối tượng Mock.

Tôi có thể thử nghiệm coroutines bằng cách nào?

class ImGoingToBeMocked: 
    @asyncio.coroutine 
    def yeah_im_not_going_to_run(self): 
     yield from asyncio.sleep(1) 
     return "sup" 

class ImBeingTested: 
    def __init__(self, hidude): 
     self.hidude = hidude 

    @asyncio.coroutine 
    def i_call_other_coroutines(self): 
     return (yield from self.hidude.yeah_im_not_going_to_run()) 

class TestImBeingTested(unittest.TestCase): 

    def test_i_call_other_coroutines(self): 
     mocked = Mock(ImGoingToBeMocked) 
     ibt = ImBeingTested(mocked) 

     ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines()) 

Answer 1

Vì thư viện mock không hỗ trợ coroutines Tôi tạo ra coroutines giả và tự gán cho đối tượng giả. Một chút chi tiết hơn nhưng nó hoạt động.

ví dụ bạn có thể trông như thế này:

import asyncio 
import unittest 
from unittest.mock import Mock 


class ImGoingToBeMocked: 
    @asyncio.coroutine 
    def yeah_im_not_going_to_run(self): 
     yield from asyncio.sleep(1) 
     return "sup" 


class ImBeingTested: 
    def __init__(self, hidude): 
     self.hidude = hidude 

    @asyncio.coroutine 
    def i_call_other_coroutines(self): 
     return (yield from self.hidude.yeah_im_not_going_to_run()) 


class TestImBeingTested(unittest.TestCase): 

    def test_i_call_other_coroutines(self): 
     mocked = Mock(ImGoingToBeMocked) 
     ibt = ImBeingTested(mocked) 

     @asyncio.coroutine 
     def mock_coro(): 
      return "sup" 
     mocked.yeah_im_not_going_to_run = mock_coro 

     ret = asyncio.get_event_loop().run_until_complete(
      ibt.i_call_other_coroutines()) 
     self.assertEqual("sup", ret) 


if __name__ == '__main__': 
    unittest.main() 

Answer 2

mọc tắt của Andrew Svetlov của answer, tôi chỉ muốn chia sẻ chức năng helper này:

def get_mock_coro(return_value): 
    @asyncio.coroutine 
    def mock_coro(*args, **kwargs): 
     return return_value 

    return Mock(wraps=mock_coro) 

này cho phép bạn sử dụng các tiêu chuẩn assert_called_with, call_count và các phương thức và thuộc tính khác là unittest.Mock thông thường cung cấp cho bạn.

Bạn có thể sử dụng với mã trong câu hỏi như:

class ImGoingToBeMocked: 
    @asyncio.coroutine 
    def yeah_im_not_going_to_run(self): 
     yield from asyncio.sleep(1) 
     return "sup" 

class ImBeingTested: 
    def __init__(self, hidude): 
     self.hidude = hidude 

    @asyncio.coroutine 
    def i_call_other_coroutines(self): 
     return (yield from self.hidude.yeah_im_not_going_to_run()) 

class TestImBeingTested(unittest.TestCase): 

    def test_i_call_other_coroutines(self): 
     mocked = Mock(ImGoingToBeMocked) 
     mocked.yeah_im_not_going_to_run = get_mock_coro() 
     ibt = ImBeingTested(mocked) 

     ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines()) 
     self.assertEqual(mocked.yeah_im_not_going_to_run.call_count, 1) 

Answer 3

Dustin của câu trả lời có lẽ là một trong những quyền trong đại đa số các trường hợp. Tôi đã có một vấn đề khác, nơi coroutine cần trả lại nhiều giá trị, ví dụ: mô phỏng hoạt động read(), như mô tả ngắn gọn trong số comment của tôi.

Sau một số thử nghiệm nhiều hơn, mã dưới đây làm việc cho tôi, bằng cách định nghĩa một iterator ngoài chức năng mocking, nhớ một cách hiệu quả các giá trị cuối cùng trở lại để gửi tới một:

def test_some_read_operation(self): 
    #... 
    data = iter([b'data', b'']) 
    @asyncio.coroutine 
    def read(*args): 
     return next(data) 
    mocked.read = Mock(wraps=read) 
    # Here, the business class would use its .read() method which 
    # would first read 4 bytes of data, and then no data 
    # on its second read. 

Vì vậy, mở rộng trên câu trả lời Dustin, nó sẽ trông như thế:

def get_mock_coro(return_values): 
    values = iter(return_values) 
    @asyncio.coroutine 
    def mock_coro(*args, **kwargs): 
     return next(values) 

    return Mock(wraps=mock_coro) 

hai nhược điểm trước mắt tôi có thể nhìn thấy trong phương pháp này là:

1. Nó không cho phép tăng ngoại lệ một cách dễ dàng (ví dụ: đầu tiên trả về một số dữ liệu, sau đó tăng lỗi trên thao tác đọc thứ hai).
2. Tôi chưa tìm thấy cách sử dụng các thuộc tính tiêu chuẩn Mock.side_effect hoặc .return_value để làm cho nó rõ ràng và dễ đọc hơn.

Answer 4

Tôi đang viết một trình bao bọc để unittest nhằm mục đích cắt giảm bản mẫu khi viết bài kiểm tra cho asyncio.

Mã này sống ở đây: https://github.com/Martiusweb/asynctest

Bạn có thể thử một coroutine với asynctest.CoroutineMock:

>>> mock = CoroutineMock(return_value='a result') 
>>> asyncio.iscoroutinefunction(mock) 
True 
>>> asyncio.iscoroutine(mock()) 
True 
>>> asyncio.run_until_complete(mock()) 
'a result' 

Nó cũng làm việc với các thuộc tính side_effect, và một asynctest.Mock với một spec có thể trở lại CoroutineMock:

>>> asyncio.iscoroutinefunction(Foo().coroutine) 
True 
>>> asyncio.iscoroutinefunction(Foo().function) 
False 
>>> asynctest.Mock(spec=Foo()).coroutine 
<class 'asynctest.mock.CoroutineMock'> 
>>> asynctest.Mock(spec=Foo()).function 
<class 'asynctest.mock.Mock'> 

Tất cả các tính năng của unittest.Mock được mong đợi ed để hoạt động chính xác (patch(), v.v.).

Answer 5

Bạn có thể tạo không đồng bộ chế giễu bản thân:

import asyncio 
from unittest.mock import Mock 


class AsyncMock(Mock): 

    def __call__(self, *args, **kwargs): 
     sup = super(AsyncMock, self) 
     async def coro(): 
      return sup.__call__(*args, **kwargs) 
     return coro() 

    def __await__(self): 
     return self().__await__()