Appel simulé d'un appel asynchrone en python 3.5

Question

Comment simuler un appel asynchrone d'une coroutine native vers une autre en utilisant unittest.mock.patch?

J'ai actuellement tout à fait une solution maladroite:

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

Puis

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

Cela fonctionne, mais semble laid. Y a-t-il une façon plus pythonique de faire ça?

Answer 1

Sous-classe MagicMock propagera votre classe personnalisée pour tous les simulacres générés à partir de votre simulacre coroutine. Par exemple, AsyncMock().__str__ deviendra également un AsyncMock ce qui n'est probablement pas ce que vous cherchez. Au lieu de cela, vous pouvez définir une fabrique qui crée un Mock (ou un MagicMock) avec des arguments personnalisés, par exemple side_effect=coroutine(coro). En outre, il peut être judicieux de séparer la fonction coroutine de la coroutine (comme expliqué dans la section documentation).

Voici ce que je suis venu avec:

from asyncio import coroutine 

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

Une explication des différents objets:

• corofunc: la fonction coroutine maquette
• corofunc.side_effect(): le coroutine, généré pour chaque appel
• corofunc.coro: le faux utilisé par la coroutine pour obtenir le résultat
• corofunc.coro.return_value: la valeur retournée par la coroutine
• corofunc.coro.side_effect: peut être utilisé pour soulever une exception

Exemple:

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

Une autre façon de se moquer de la coroutine est de faire de la coroutine, qui renvoie faux. De cette façon, vous pouvez simuler des coroutines qui seront passées en asyncio.wait ou asyncio.wait_for.

Cela rend plus coroutines universelle fait que la configuration des tests plus encombrants:

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

La solution est en fait assez simple: J'ai juste besoin de convertir la méthode __call__ de la maquette en coroutine:

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

Cela fonctionne parfaitement, quand un simulacre est appelé, le code reçoit la coroutine natif

Answer 4

Tout le monde est absent ce qui est probablement la plus simple et la plus claire solution:

@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) 

rappelez-vous qu'une coroutine peut être pensée de juste comme une fonction qui est garantie pour retourner un avenir qui peut, à son tour être attendu.