Come deridere le coroutine asyncio?

Question

Il seguente codice non funziona con TypeError: 'Mock' object is not iterable in ImBeingTested.i_call_other_coroutines perché ho sostituito ImGoingToBeMocked da un oggetto Mock.

Come posso deridere le coroutine?

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

Da mock libreria non supporta coroutine Creo coroutine deriso manualmente e assegnare quelle oggetto fittizio. Un po 'più prolisso ma funziona.

Il vostro esempio può apparire come segue:

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

Molleggio off di di answer Andrew Svetlov, volevo solo condividere questa funzione di supporto:

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

    return Mock(wraps=mock_coro) 

Ciò consente di utilizzare lo standard assert_called_with, call_count e altri metodi e attribuisce un unittest regolare. Ti dà.

È possibile utilizzare questo con il codice nella domanda del tipo:

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

risposta di Dustin è probabilmente quella giusta nella stragrande maggioranza dei casi. Ho avuto un problema diverso in cui la coroutine doveva restituire più di un valore, ad es. simulare un'operazione di read(), come descritto brevemente nel mio comment.

Dopo un po 'di più test, il codice qui sotto ha lavorato per me, attraverso la definizione di un iteratore di fuori della funzione beffardo, ricordando in modo efficace l'ultimo valore restituito per inviare la prossima:

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. 

Quindi, ampliando la risposta di Dustin, sarebbe simile:

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) 

I due aspetti negativi immediati posso vedere in questo approccio sono:

1. Non consente di aumentare facilmente le eccezioni (ad es. prima restituisci alcuni dati, quindi genera un errore sull'operazione di seconda lettura).
2. Non ho trovato un modo per utilizzare gli attributi standard Mock.side_effect o .return_value per renderlo più ovvio e leggibile.

Answer 4

Sto scrivendo un wrapper per unittest che mira a tagliare il boilerplate durante la scrittura di test per asyncio.

Il codice vive qui: https://github.com/Martiusweb/asynctest

È possibile prendere in giro un coroutine con asynctest.CoroutineMock:

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

Funziona anche con l'attributo side_effect, ed un asynctest.Mock con un spec può restituire 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'> 

Tutte le funzionalità di unittest.Mock sono previste ed a funzionare correttamente (patch(), ecc.).

Answer 5

È possibile creare asincrona prende in giro se stessi:

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__()