Synnefo » snf-ganeti » Ganeti ExtStorage providers

#!/usr/bin/env python

#

# Copyright (c) 2010-2013 Greek Research and Technology Network S.A.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

try:

    import unittest2 as unittest

except ImportError:

    import unittest

import os

from mock import patch

from storwize_svc import *

import subprocess

@patch('paramiko.SSHClient', autospec=True)

class SVCConnectionTest(unittest.TestCase):

    def test_ssh_injection_1(self, ssh_conn):

        self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass')

        cmd_list = ['ls', '- l']

        self.assertRaises(SVCError, self.conn.run_ssh_cmd, cmd_list)

    def test_ssh_injection_2(self, ssh_conn):

        self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass')

        cmd_list = ['ls', ';cat>/dev/sda</dev/zero']

        self.assertRaises(SVCError, self.conn.run_ssh_cmd, cmd_list)

    @patch('paramiko.ChannelFile')

    def test_initial_connection_ok(self, chan, ssh_conn):

        self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass')

        cmd_list = ['ls', '-l']

        chan.channel.recv_exit_status.return_value = 0

        ssh_conn.return_value.exec_command.return_value = (chan, chan, chan)

        self.conn.run_ssh_cmd(cmd_list)

        kwargs = {'port': 22,

                  'username': 'user',

                  'password': 'userpass',

                  'timeout': 30,

                  'look_for_keys': False,

                  'allow_agent': False}

        ssh_conn.return_value.connect.assert_called_once_with('127.0.0.1',

                                                              **kwargs)

    @patch('paramiko.ChannelFile')

    def test_initial_connection_pooled(self, chan, ssh_conn):

        self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass')

        cmd_list = ['ls', '-l']

        chan.channel.recv_exit_status.return_value = 0

        ssh_conn.return_value.exec_command.return_value = (chan, chan, chan)

        self.conn.run_ssh_cmd(cmd_list)

        self.conn.run_ssh_cmd(cmd_list)

        kwargs = {'port': 22,

                  'username': 'user',

                  'password': 'userpass',

                  'timeout': 30,

                  'look_for_keys': False,

                  'allow_agent': False}

        ssh_conn.return_value.connect.assert_called_once_with('127.0.0.1',

                                                              **kwargs)

    @patch('paramiko.ChannelFile')

    def test_initial_connection_fail(self, chan, ssh_conn):

        self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass')

        cmd_list = ['ls', '-l']

        chan.channel.recv_exit_status.return_value = -1

        ssh_conn.return_value.exec_command.return_value = (chan, chan, chan)

        self.assertRaises(SVCError, self.conn.run_ssh_cmd, cmd_list)

    @patch('paramiko.ChannelFile')

    def test_initial_connection_out(self, chan, ssh_conn):

        self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass')

        cmd_list = ['ls', '-l']

        chan.channel.recv_exit_status.return_value = 0

        chan.read.return_value = "test"

        ssh_conn.return_value.exec_command.return_value = (chan, chan, chan)

        out, err = self.conn.run_ssh_cmd(cmd_list)

        self.assertEqual(out, "test")

        self.assertEqual(err, "test")

    @patch('paramiko.ChannelFile')

    def test_initial_connection_pooled_retry(self, chan, ssh_conn):

        self.conn = SVCConnection('127.0.0.1', 22, 'user', 'userpass')

        cmd_list = ['ls', '-l']

        chan.channel.recv_exit_status.return_value = 0

        ssh_conn.return_value.exec_command.return_value = (chan, chan, chan)

        self.conn.run_ssh_cmd(cmd_list)

        transport = ssh_conn.return_value.get_transport.return_value

        transport.is_active.return_value = False

        self.conn.run_ssh_cmd(cmd_list)

        self.assertEqual(ssh_conn.return_value.connect.call_count, 2)

@patch('storwize_svc.SVCConnection', autospec=True)

class SVCInitTest(unittest.TestCase):

    def test_missing_config(self, svc_conn):

        self.assertRaises(TypeError, SVCProvider)

    def test_missing_env(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './test_configs/some_config')

    @patch.dict(os.environ, {'VOL_SIZE': '1', 'EXTP_SVC_SAN': 'mysan'})

    def test_missing_vol_name(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './test_configs/some_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '1'})

    def test_missing_san(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './test_configs/some_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '1',

                             'EXTP_SVC_SAN': 'mysan'})

    def test_bad_config_path(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './test_configs/some_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'EXTP_SVC_SAN': 'mysan'})

    def test_bad_identifier(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './test_configs/good_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_missing_size(self, svc_conn):

        svc = SVCProvider('./test_configs/good_config')

        self.assertEqual(svc._vol_name, 'myvol')

        self.assertIsNone(svc._vol_size)

        self.assertEqual(svc._svc_pool, 'VM_POOL_SYNNEFO')

        self.assertEqual(svc._host_id, 2)

        svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser',

                                         'uberpass')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_good_config(self, svc_conn):

        svc = SVCProvider('./test_configs/good_config')

        self.assertEqual(svc._vol_name, 'myvol')

        self.assertEqual(svc._vol_size, 100)

        self.assertEqual(svc._svc_pool, 'VM_POOL_SYNNEFO')

        self.assertEqual(svc._host_id, 2)

        svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser',

                                         'uberpass')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_good_config_extra_sections(self, svc_conn):

        svc = SVCProvider('./test_configs/good_config_extra_sections')

        self.assertEqual(svc._vol_name, 'myvol')

        self.assertEqual(svc._vol_size, 100)

        self.assertEqual(svc._svc_pool, 'synnefo')

        self.assertEqual(svc._host_id, 2)

        svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser',

                                         'uberpass')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_good_config_extra_empty_sections(self, svc_conn):

        svc = SVCProvider('./test_configs/good_config_extra_empty_sections')

        self.assertEqual(svc._vol_name, 'myvol')

        self.assertEqual(svc._vol_size, 100)

        self.assertEqual(svc._svc_pool, 'synnefo')

        self.assertEqual(svc._host_id, 2)

        svc_conn.assert_called_once_with('10.0.2.1', 22, 'myuser',

                                         'uberpass')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_bad_config_section(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider,

                          './test_configs/bad_config_section')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_bad_config_missing_section(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider,

                          './test_configs/bad_config_missing_section')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_bad_config_missing_fields(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider,

                          './test_configs/bad_config_missing_fields')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_bad_config_invalid_format(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider,

                          './test_configs/bad_config_invalid_format')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_bad_config_empty_host(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider,

                          './test_configs/bad_config_empty_host')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_bad_config_bad_port(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider,

                          './test_configs/bad_config_bad_port')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_bad_config_bad_hostid(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider,

                          './test_configs/bad_config_bad_hostid')

    @patch.dict(os.environ, {'VOL_NAME': '', 'VOL_SIZE': '1',

                             'EXTP_SVC_SAN': 'mysan'})

    def test_empty_volume_name(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './some_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '',

                             'EXTP_SVC_SAN': 'mysan'})

    def test_empty_volume_size(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './some_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': 'test',

                             'EXTP_SVC_SAN': 'mysan'})

    def test_bad_volume_size(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './some_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '-1',

                             'EXTP_SVC_SAN': 'mysan'})

    def test_negative_volume_size(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './some_config')

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '-1',

                             'EXTP_SVC_SAN': ''})

    def test_empty_svc_identifier(self, svc_conn):

        self.assertRaises(SVCError, SVCProvider, './some_config')

class SVCEnvOpsTest(unittest.TestCase):

    @patch.dict(os.environ, {'VOL_NAME': 'myvol',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_create_no_size(self):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc.create)

    @patch.dict(os.environ, {'VOL_NAME': 'myvol',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_grow_no_size(self):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc.grow)

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_detach_size(self):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc.detach)

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_attach_size(self):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc.attach)

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_verify_size(self):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc.verify)

    @patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                             'EXTP_SVC_SAN': 'svc_san_identifier'})

    def test_remove_size(self):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc.remove)

@patch('subprocess.check_output', autospec=True)

@patch('subprocess.check_call', autospec=True)

@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                         'EXTP_SVC_SAN': 'svc_san_identifier'})

class SVCInternalHostOpsTest(unittest.TestCase):

    def test_check_call_ok(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        svc._check_call(['ls', '-l'])

    def test_check_call_exc(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        call.side_effect = subprocess.CalledProcessError(1, "test",

                                                         output="test")

        self.assertRaises(SVCError, svc._check_call, ['ls', '-l'])

    def test_check_output_ok(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = "test"

        ret = svc._check_output(['ls', '-l'])

        self.assertEqual(ret, "test")

    def test_check_output_empty(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = ""

        ret = svc._check_output(['ls', '-l'])

        self.assertEqual(ret, "")

    def test_check_output_exc(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = "test"

        out.side_effect = subprocess.CalledProcessError(1, "test",

                                                        output="test")

        self.assertRaises(SVCError, svc._check_output, ['ls', '-l'])

    @patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc",

                                      "/dev/sdd"], autospec=True)

    def test_devices_add_ok(self, glob, call, out):

        svc = SVCProvider('./test_configs/good_config')

        svc._devices_add(12)

    @patch('glob.glob', return_value=[], autospec=True)

    def test_devices_add_not_discovered(self, glob, call, out):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc._devices_add, 12)

    @patch('glob.glob', return_value=["/dev/sda", "/dev/sdc"], autospec=True)

    def test_devices_add_partial_discovered(self, glob, call, out):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc._devices_add, 12)

    @patch('glob.glob', return_value=["/dev/sda", "/dev/sdc"], autospec=True)

    def test_devices_add_scan_exc(self, glob, call, out):

        svc = SVCProvider('./test_configs/good_config')

        call.side_effect = SVCError("failed")

        self.assertRaises(SVCError, svc._devices_add, 12)

    @patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc",

                                      "/dev/sdd"], autospec=True)

    @patch.object(SVCProvider, '_devices_discovered', return_value=False,

                  autospec=True)

    def test_devices_remove_ok(self, glob, disc, call, out):

        svc = SVCProvider('./test_configs/good_config')

        with patch('__builtin__.open', autospec=True):

            svc._devices_remove(12)

    @patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc",

                                      "/dev/sdd"], autospec=True)

    @patch.object(SVCProvider, '_devices_discovered', return_value=True,

                  autospec=True)

    def test_devices_remove_fail(self, glob, disc, call, out):

        svc = SVCProvider('./test_configs/good_config')

        with patch('__builtin__.open', autospec=True):

            self.assertRaises(SVCError, svc._devices_remove, 12)

    @patch('glob.glob', return_value=["/dev/sda", "/dev/sdb", "/dev/sdc",

                                      "/dev/sdd"], autospec=True)

    @patch.object(SVCProvider, '_devices_discovered', return_value=True,

                  autospec=True)

    def test_devices_remove_exc(self, glob, disc, call, out):

        svc = SVCProvider('./test_configs/good_config')

        with patch('__builtin__.open', autospec=True) as op:

            op.side_effect = IOError("foo")

            self.assertRaises(SVCError, svc._devices_remove, 12)

    def test_multipath_ok_ok(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = "multipath -ll command output not empty"

        ret = svc._multipath_ok("/dev/mapper/3123")

        self.assertEqual(ret, True)

    def test_multipath_ok_fail(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = ""

        ret = svc._multipath_ok("/dev/mapper/3123")

        self.assertEqual(ret, False)

    def test_multipath_ok_exc(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        call.side_effect = SVCError("command failed")

        self.assertRaises(SVCError, svc._multipath_ok, "/dev/mapper/3123")

    def test_multipath_setup_ok(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = "multipath -ll command output not empty"

        svc._multipath_setup("/dev/mapper/3123")

    def test_multipath_setup_fail(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = ""

        self.assertRaises(SVCError, svc._multipath_setup, "/dev/mapper/3123")

    def test_multipath_setup_exc(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        call.side_effect = SVCError("command failed")

        self.assertRaises(SVCError, svc._multipath_setup, "/dev/mapper/3123")

    def test_multipath_remove_ok(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = ""

        svc._multipath_remove("/dev/mapper/3123")

    def test_multipath_remove_fail(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        out.return_value = "multipath -ll cmd out"

        self.assertRaises(SVCError, svc._multipath_remove, "/dev/mapper/3123")

    def test_multipath_remove_exc(self, call, out):

        svc = SVCProvider('./test_configs/good_config')

        call.side_effect = SVCError("command failed")

        self.assertRaises(SVCError, svc._multipath_remove, "/dev/mapper/3123")

    #TODO: host_device_setup/remove , fix testing for devices_* methods, rename

    # devices to device

@patch.object(SVCConnection, 'run_ssh_cmd', autospec=True)

@patch.dict(os.environ, {'VOL_NAME': 'myvol', 'VOL_SIZE': '100',

                         'EXTP_SVC_SAN': 'svc_san_identifier'})

class SVCInternalRemoteOpsTest(unittest.TestCase):

    def test_get_device_info_ok(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "vdisk_UID!1\nSCSI_id!2\ntest3!3\ntest4!4\n"

        run_ssh_cmd.return_value = (ret_str, "")

        ret = svc._get_device_info('myvol')

        ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol']

        run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd)

        self.assertEqual(run_ssh_cmd.call_count, 2)

        exp_ret = ("/dev/mapper/31", 2)

        self.assertTupleEqual(ret, exp_ret)

    def test_get_device_info_malformed(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "vdisk_UID!ab\nSCSI_id!2\ntest3!3\ntest4!4\n"

        run_ssh_cmd.return_value = (ret_str, "")

        self.assertRaises(SVCError, svc._get_device_info, 'myvol')

        ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol']

        run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd)

        self.assertEqual(run_ssh_cmd.call_count, 2)

    def test_get_device_info_missing(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "vdisk_UID!ab\nsome_field!2\ntest3!3\ntest4!4\n"

        run_ssh_cmd.return_value = (ret_str, "")

        self.assertRaises(SVCError, svc._get_device_info, 'myvol')

        ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol']

        run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd)

        self.assertEqual(run_ssh_cmd.call_count, 2)

    def test_get_device_info_empty(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("", "")

        self.assertRaises(SVCError, svc._get_device_info, 'myvol')

        ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol']

        run_ssh_cmd.assert_called_with(svc._svc_connection, ssh_cmd)

        self.assertEqual(run_ssh_cmd.call_count, 1)

    def test_vdisk_defined(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "test1!1\ntest2!2\ntest3!3\ntest4!4\n"

        run_ssh_cmd.return_value = (ret_str, "")

        ret = svc._is_vdisk_defined('myvol')

        ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol']

        run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd)

        self.assertEqual(ret, True)

    #FIXME:

    def test_vdisk_not_defined(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("", "")

        ret = svc._is_vdisk_defined('myvol')

        ssh_cmd = ['svcinfo', 'lsvdisk', '-bytes', '-delim', '!', 'myvol']

        run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd)

        self.assertEqual(ret, False)

    def test_driver_assert_ok(self, run_ssh_cmd):

        SVCProvider._driver_assert(1, 'ok')

    def test_driver_assert_fail(self, run_ssh_cmd):

        self.assertRaises(SVCError, SVCProvider._driver_assert, 0, 'fail')

    def test_assert_ssh_return_ok(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        svc._assert_ssh_return(1, 'fun', ['ls', '-l'], 'test', 'test')

    def test_assert_ssh_return_fail(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(SVCError, svc._assert_ssh_return, 0,

                          'fun', ['ls', '-l'], 'test', 'test')

    def test_exec_cmd_parse_ok(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "test1!1\ntest2!2\ntest3!3\ntest4!4\n"

        run_ssh_cmd.return_value = (ret_str, "")

        ret = svc._exec_cmd_and_parse_attrs('ssh_cmd placeholder')

        ret_dict = {'test1': '1', 'test2': '2', 'test3': '3', 'test4': '4'}

        self.assertDictEqual(ret, ret_dict)

    def test_exec_cmd_parse_empty(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("", "")

        self.assertRaises(SVCError, svc._exec_cmd_and_parse_attrs,

                          'ssh_cmd placeholder')

    def test_exec_cmd_parse_malformed(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "test11 a\ntest2=2\ntest3\ntest4  test\n"

        run_ssh_cmd.return_value = (ret_str, "")

        ret = svc._exec_cmd_and_parse_attrs('ssh_cmd placeholder')

        ret_dict = {'test11 a': '', 'test2=2': '', 'test3': '',

                    'test4  test': ''}

        self.assertDictEqual(ret, ret_dict)

    def test_get_vdisk_attributes_ok(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "test1!1\ntest2!2\ntest3!3\ntest4!4\n"

        run_ssh_cmd.return_value = (ret_str, "")

        ret = svc._get_vdisk_attributes('ssh_cmd placeholder')

        ret_dict = {'test1': '1', 'test2': '2', 'test3': '3', 'test4': '4'}

        self.assertDictEqual(ret, ret_dict)

    def test_get_vdisk_attributes_empty(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("", "")

        self.assertRaises(SVCError, svc._get_vdisk_attributes,

                          'ssh_cmd placeholder')

    def test_get_vdisk_attributes_malformed(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret_str = "test11 a\ntest2=2\ntest3\ntest4  test\n"

        run_ssh_cmd.return_value = (ret_str, "")

        ret = svc._get_vdisk_attributes('ssh_cmd placeholder')

        ret_dict = {'test11 a': '', 'test2=2': '', 'test3': '',

                    'test4  test': ''}

        self.assertDictEqual(ret, ret_dict)

        self.assertDictEqual(ret, ret_dict)

    def test_get_hdr_dic_ok(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        hdr_str = "field1!field2!field3!field4"

        row_str = "test1!test2!test3!test4"

        ret = svc._get_hdr_dic(hdr_str, row_str, '!')

        ret_dict = dict(zip(hdr_str.split('!'), row_str.split('!')))

        self.assertDictEqual(ret, ret_dict)

    def test_get_hdr_dic_malformed(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        hdr_str = "field1!field2!field3!field4"

        row_str = "test1!test2!test3"

        self.assertRaises(SVCError, svc._get_hdr_dic, hdr_str, row_str, '!')

    def test_get_hdr_dic_empty(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        ret = svc._get_hdr_dic("", "", '!')

        self.assertDictEqual(ret, {"": ""})

    def test_get_hostvdisk_maps_ok(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        hdr_str = "field1!vdisk_name!field3!field4\n"

        row1_str = "test1!test2!test3!test4\n"

        row2_str = "test11!test12!test13!test14\n"

        ret_str = hdr_str + row1_str + row2_str

        run_ssh_cmd.return_value = (ret_str, "")

        ret = svc._get_hostvdisk_mappings(5)

        ssh_cmd = ['svcinfo', 'lshostvdiskmap', '-delim', '!', '5']

        run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd)

        row1_dict = {'field1': 'test1',

                     'vdisk_name': 'test2',

                     'field3': 'test3',

                     'field4': 'test4'}

        row2_dict = {'field1': 'test11',

                     'vdisk_name': 'test12',

                     'field3': 'test13',

                     'field4': 'test14'}

        ret_dict = {'test2': row1_dict, 'test12': row2_dict}

        self.assertDictEqual(ret, ret_dict)

    def test_get_hostvdisk_maps_malformed(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        hdr_str = "field1!field2!field3!field4\n"

        row1_str = "test1!test2!test3!test4\n"

        row2_str = "test11!test12!test13!test14\n"

        ret_str = hdr_str + row1_str + row2_str

        run_ssh_cmd.return_value = (ret_str, "")

        self.assertRaises(Exception, svc._get_hostvdisk_mappings, 5)

    def test_get_hostvdisk_maps_empty(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("", "")

        ret = svc._get_hostvdisk_mappings(5)

        self.assertDictEqual(ret, {})

    def test_get_hostvdisk_maps_only_header(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        hdr_str = "field1!vdisk_name!field3!field4\n"

        run_ssh_cmd.return_value = (hdr_str, "")

        ret = svc._get_hostvdisk_mappings(5)

        self.assertDictEqual(ret, {})

    def test_get_hostvdisk_maps_exc(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.side_effect = SVCError("Cannot run ssh cmd")

        self.assertRaises(SVCError, svc._get_hostvdisk_mappings, 5)

    def test_map_vol_to_host(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("successfully created vol", "")

        svc._map_vol_to_host('myvol', 7)

        ssh_cmd = ['svctask', 'mkvdiskhostmap', '-host', '7', 'myvol']

        run_ssh_cmd.assert_called_once_with(svc._svc_connection, ssh_cmd)

    def test_map_vol_to_host_malformed(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("malformed output", "")

        self.assertRaises(SVCError, svc._map_vol_to_host, 'myvol', 7)

    def test_map_vol_to_host_empty(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        run_ssh_cmd.return_value = ("", "")

        self.assertRaises(SVCError, svc._map_vol_to_host, 'myvol', 7)

    def test_map_vol_to_host_assert(self, run_ssh_cmd):

        svc = SVCProvider('./test_configs/good_config')

        self.assertRaises(AssertionError, svc._map_vol_to_host, 'myvol', 5)