test_client.py 文件源码

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

    client._ws_connection = mock_ws
    client._ws_session_data = {}
    client._handle_event = MagicMock()

    client.loop.run_until_complete(client.ws_close())

    assert mock_ws.close.called
    assert len(mock_ws.close.mock_calls) == 1
    assert mock_ws.send_str.called
    assert len(mock_ws.send_str.mock_calls) == 1
    assert mock_ws.send_str.mock_calls[0][1][0] == '41'