在Python 3.5中模拟异步调用

69 人关注

如何使用 unittest.mock.patch 模拟从一个本地程序到另一个程序的异步调用？

我目前有一个相当尴尬的解决方案。

class CoroutineMock(MagicMock):
    def __await__(self, *args, **kwargs):
        future = Future()
        future.set_result(self)
        result = yield from future
        return result
class TestCoroutines(TestCase):
    @patch('some.path', new_callable=CoroutineMock)
    def test(self, mock):
        some_action()
        mock.assert_called_with(1,2,3)
这样做可以，但看起来很丑。是否有更多的pythonic方法来做这个？


           
            
             另外，由于asyncio.tasks.sure_future的原因，这个模拟不能与asyncio.await一起工作。


         
          python


         
          python-asyncio


         
          python-mock


          
           已采纳


          
           
            每个人都错过了可能是最简单和最清晰的解决方案。
           
           @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)
请记住，coroutine可以被认为是一个保证返回未来的函数，而这个未来又可以被等待。


           
            
             
              process_smtp_message.return_value = f是什么？  另外，对被测试的函数的调用在哪里？


           
            
             
              @Skorpeo - 我想他是指mock.return_value = f


           
            
             
              我绝对不知道。process_smtp_message显然是你要模拟的东西。


           
            
             
              Sergey Nevmerzhitsky
             
             ：


           
            
             
              该方案与Python 3.8不兼容，后者引入了一个本地
              
               AsyncMock
              
              。所以使用该解决方案的代码会因为Future类的问题而出错。但是，Zozz的解决方案与一个简单的
              
               AsyncMock
              
              的实现可以在Python 3.7（例如）和Python 3.8（如果你将有条件地导入本地
              
               AsyncMock
              
              ）中工作。


           
            
             
              Kishan Kishore
             
             ：


           
            
             
              对于
              
               python3.8
              
              及以上，我最终使用。【替换代码1


          
           
            
             解决办法其实很简单。
我只需要将mock的
             
              __call__
             
             方法转换为coroutine。
            
            class AsyncMock(MagicMock):
    async def __call__(self, *args, **kwargs):
        return super(AsyncMock, self).__call__(*args, **kwargs)
这样做效果很好，当mock被调用时，代码会收到本地的coroutine
使用实例。
@mock.patch('my.path.asyncio.sleep', new_callable=AsyncMock)
def test_stuff(sleep):
    # code


           
            
             
              
               
                这很好，但它不能与autospec一起使用，而autospec在使用MagicMock时基本上是强制性的。有什么想法可以让它工作吗？我对其内部结构还不够熟悉...


           
            
             
              
               
                它对我来说非常有效。我是这样使用它的。``` @mock.patch( 'my.path.asyncio.sleep', new_callable=AsyncMock, ) def test_stuff(sleep):     #代码 ````


           
            
             
              
               
                Ytsen de Boer
               
               ：


           
            
             
              
               
                这个办法可行。我最初喜欢下面Ivan Castellanos的另一个解决方案。但未来从来没有执行过，我拼命尝试，但没能成功。


           
            
             
              
               
                这是最方便和最优雅的。至少在基本使用方面是这样。


           
            
             
              
               
                注意。从python 3.8开始，
                
                 AsyncMock
                
                在
                
                 unittest.mock
                
                中可用，mock也会自动检测到它应该被使用的地方（见
                
                 示例的文档
                
                ).


          
           
            
             
              
               
               
                Ivan Castellanos
               
              
              
               发布于
               
               2022-05-10


          
           
            
             
              
               根据@scolvin的回答，我创造了这个（我认为）更干净的方法。
              
              import asyncio
def async_return(result):
    f = asyncio.Future()
    f.set_result(result)
    return f
就是这样，只要在你想要的异步返回周围使用它就可以了，例如
mock = MagicMock(return_value=async_return("Example return"))
await mock()


          
           
            
             
              
               
                
                 子类化
                 
                  MagicMock
                 
                 会将你的自定义类传播给所有由你的coroutine mock生成的mock。例如，
                 
                  AsyncMock().__str__
                 
                 也会变成
                 
                  AsyncMock
                 
                 ，这可能不是你想要的。
                
                
                 相反，你可能想定义一个工厂，用自定义参数创建一个
                 
                  Mock
                 
                 （或一个
                 
                  MagicMock
                 
                 ），例如
                 
                  side_effect=coroutine(coro)
                 
                 。此外，将冠状动脉函数与冠状动脉分离可能是一个好主意（如在
                 
                  文件
                 
                 ).
                
                
                 这是我想出的办法。
                
                from asyncio import coroutine
def CoroMock():
    coro = Mock(name="CoroutineResult")
    corofunc = Mock(name="CoroutineFunction", side_effect=coroutine(coro))
    corofunc.coro = coro
    return corofunc
对不同对象的解释。
corofunc: the coroutine function mock
corofunc.side_effect(): the coroutine, generated for each call
corofunc.coro: the mock used by the coroutine to get the result
corofunc.coro.return_value: the value returned by the coroutine
corofunc.coro.side_effect: might be used to raise an exception
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


           
            
             
              
               
                
                 
                  
                   
                    this doesn't work, side_effect=coroutine(coro), coroutine is not defined


           
            
             
              
               
                
                 
                  
                   
                    其实我更喜欢原来的解决方案，因为它不需要对测试函数进行特殊的重写。这种方法比问题中所示的方法有什么优势吗？


          
           
            
             
              
               
                
                 
                  
                   另一种嘲弄coroutine的方法是制作coroutine，返回mock。这样你就可以模拟那些将被传入
                   
                    asyncio.wait
                   
                   或
                   
                    asyncio.wait_for
                   
                   的coroutine。
                  
                  
                   这使得更多的通用冠词，虽然使测试的设置更加繁琐。
                  
                  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()


          
           
            
             
              
               
                
                 
                  
                   
                    
                    
                     Murphy Meng
                    
                   
                   
                    发布于
                    
                    2022-05-10


          
           
            
             
              
               
                
                 
                  
                   
                    还有一个 "最简单 "的解决方案的变种，即模拟一个异步对象，这只是一个单行本。
                   
                   
                    In source:
                   
                   class Yo:
    async def foo(self):
        await self.bar()
    async def bar(self):
        # Some code
In test:
from asyncio import coroutine
yo = Yo()
# Here bounded method bar is mocked and will return a customised result.
yo.bar = Mock(side_effect=coroutine(lambda:'the awaitable should return this'))
event_loop.run_until_complete(yo.foo())


          
           
            
             
              
               
                
                 
                  
                   
                    
                     
                      我不知道为什么没有人提到可用的默认选项。
python提供了一个Async版本的MagicMock。
                     
                     
                      你可以在这里阅读更多相关内容。
                      
                       https://docs.python.org/3/library/unittest.mock.html#unittest.mock.AsyncMock
                      
                     
                     
                      如果你使用的是补丁，那么你也不需要做任何其他改变。如果需要的话，它会自动用异步模拟函数来替换它。
在此阅读更多信息
                      
                       https://docs.python.org/3/library/unittest.mock.html#patch


           
            
             
              
               
                
                 
                  
                   
                    
                     
                      
                       Peter Halverson
                      
                      ：


           
            
             
              
               
                
                 
                  
                   
                    
                     
                      
                       也许是因为这个线程指的是Python 3.5 (以及推而广之，3.8之前的所有后续版本，那时
                       
                        AsyncMock
                       
                       被引入了。）


          
           
            
             
              
               
                
                 
                  
                   
                    
                     
                      你可以这样设置一个异步方法的
                      
                       return_value
                      
                      。
                     
                     mock = unittest.mock.MagicMock()
mock.your_async_method.return_value = task_from_result(your_return_value)
async def task_from_result(result):
    return result
调用者将不得不进行await your_async_method(..)，就像该方法没有被模拟一样。


          
           
            
             
              
               
                
                 
                  
                   
                    
                     
                      
                       
                       
                        zhukovgreen
                       
                      
                      
                       发布于
                       
                       2022-05-10


          
           
            
             
              
               
                
                 
                  
                   
                    
                     
                      
                       我喜欢这种方法，这也使得AsyncMock的行为与Mock完全一样。
                      
                      class AsyncMock:
    def __init__(self, *args, **kwargs):
        self.mock = Mock(*args, **kwargs)
    async def __call__(self, *args, **kwargs):
        return self.mock(*args, **kwargs)
    def __getattr__(self, item):
        return getattr(self.mock, item)
然后你可以用与Mock相同的方式来处理它，即
@pytest.mark.asyncio
async def test_async_mock_example(monkeypatch):