test_client.py 文件源码

def test_ws_loop_exception(client):
    """Test websocket loop exception."""
    @asyncio.coroutine
    def side_effect(*args, **kwargs):
        raise aiohttp.ClientError("Mock Exception")
    mock_ws = AsyncMock()
    mock_ws.receive.side_effect = side_effect

    client._ws_connection = mock_ws
    client.ws_close = AsyncMock()
    client._handle_event = MagicMock()

    with pytest.raises(exceptions.TransportError):
        client.loop.run_until_complete(client._ws_loop())

    assert client.ws_close.called
    assert len(client.ws_close.mock_calls) == 1
    assert client._handle_event.called
    assert len(client._handle_event.mock_calls) == 1
    assert client._handle_event.mock_calls[0][1][0] == 'closed'
    assert client._handle_event.mock_calls[0][1][1] is None