# -*- coding: utf-8 -*- ''' :codeauthor: Nitin Madhok ` tests.unit.modules.zpool_test ~~~~~~~~~~~~~~~~~~~~~~~~~~~ ''' # Import Python libs from __future__ import absolute_import # Import Salt Testing libs from tests.support.mixins import LoaderModuleMockMixin from tests.support.unit import skipIf, TestCase from tests.support.mock import ( Mock, MagicMock, patch, NO_MOCK, NO_MOCK_REASON, ) # Import Salt Execution module to test import salt.modules.zpool as zpool # Import Salt Utils from salt.utils.odict import OrderedDict # Skip this test case if we don't have access to mock! @skipIf(NO_MOCK, NO_MOCK_REASON) class ZpoolTestCase(TestCase, LoaderModuleMockMixin): ''' This class contains a set of functions that test salt.modules.zpool module ''' def setup_loader_modules(self): patcher = patch('salt.modules.zpool._check_zpool', MagicMock(return_value='/sbin/zpool')) patcher.start() self.addCleanup(patcher.stop) return {zpool: {}} def test_exists_success(self): ''' Tests successful return of exists function ''' ret = {} ret['stdout'] = "NAME SIZE ALLOC FREE CAP DEDUP HEALTH ALTROOT\nmyzpool 149G 128K 149G 0% 1.00x ONLINE -" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): self.assertTrue(zpool.exists('myzpool')) def test_exists_failure(self): ''' Tests failure return of exists function ''' ret = {} ret['stdout'] = "" ret['stderr'] = "cannot open 'myzpool': no such pool" ret['retcode'] = 1 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): self.assertFalse(zpool.exists('myzpool')) def test_healthy(self): ''' Tests successful return of healthy function ''' ret = {} ret['stdout'] = "all pools are healthy" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): self.assertTrue(zpool.healthy()) def test_status(self): ''' Tests successful return of status function ''' ret = {} ret['stdout'] = "\n".join([ " pool: mypool", " state: ONLINE", " scan: scrub repaired 0 in 0h6m with 0 errors on Mon Dec 21 02:06:17 2015", "config:", "", " NAME STATE READ WRITE CKSUM", " mypool ONLINE 0 0 0", " mirror-0 ONLINE 0 0 0", " c2t0d0 ONLINE 0 0 0", " c2t1d0 ONLINE 0 0 0", "", "errors: No known data errors", ]) ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.status() self.assertEqual('ONLINE', ret['mypool']['state']) def test_iostat(self): ''' Tests successful return of iostat function ''' ret = {} ret['stdout'] = "\n".join([ " capacity operations bandwidth", "pool alloc free read write read write", "---------- ----- ----- ----- ----- ----- -----", "mypool 46.7G 64.3G 4 19 113K 331K", " mirror 46.7G 64.3G 4 19 113K 331K", " c2t0d0 - - 1 10 114K 334K", " c2t1d0 - - 1 10 114K 334K", "---------- ----- ----- ----- ----- ----- -----", ]) ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.iostat('mypool') self.assertEqual('46.7G', ret['mypool']['mypool']['capacity-alloc']) def test_list(self): ''' Tests successful return of list function ''' ret = {} ret['stdout'] = "mypool\t1.81T\t714G\t1.11T\t38%\tONLINE" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}), \ patch('salt.modules.zpool._check_features', MagicMock(return_value=False)): ret = zpool.list_() res = OrderedDict([('mypool', {'alloc': '714G', 'cap': '38%', 'free': '1.11T', 'health': 'ONLINE', 'size': '1.81T'})]) self.assertEqual(res, ret) def test_list_parsable(self): ''' Tests successful return of list function with parsable output ''' ret = {} ret['stdout'] = "mypool\t1992864825344\t767076794368\t1225788030976\t38\tONLINE" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}), \ patch('salt.modules.zpool._check_features', MagicMock(return_value=False)): ret = zpool.list_() res = OrderedDict([('mypool', {'alloc': 767076794368, 'cap': 38, 'free': 1225788030976, 'health': 'ONLINE', 'size': 1992864825344})]) self.assertEqual(res, ret) def test_get(self): ''' Tests successful return of get function ''' ret = {} ret['stdout'] = "size\t1.81T\t-\n" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.get('mypool', 'size') res = OrderedDict([('mypool', OrderedDict([('size', '1.81T')]))]) self.assertEqual(res, ret) def test_get_parsable(self): ''' Tests successful return of get function with parsable output ''' ret = {} ret['stdout'] = "size\t1992864825344\t-\n" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.get('mypool', 'size') res = OrderedDict([('mypool', OrderedDict([('size', 1992864825344)]))]) self.assertEqual(res, ret) def test_get_whitespace(self): ''' Tests successful return of get function with a string with whitespaces ''' ret = {} ret['stdout'] = "comment\tmy testing pool\t-\n" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.get('mypool', 'comment') res = OrderedDict([('mypool', OrderedDict([('comment', "'my testing pool'")]))]) self.assertEqual(res, ret) def test_scrub_start(self): ''' Tests start of scrub ''' ret = {} ret['stdout'] = "" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) mock_exists = MagicMock(return_value=True) with patch.dict(zpool.__salt__, {'zpool.exists': mock_exists}): with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.scrub('mypool') res = OrderedDict([('mypool', OrderedDict([('scrubbing', True)]))]) self.assertEqual(res, ret) def test_scrub_pause(self): ''' Tests pause of scrub ''' ret = {} ret['stdout'] = "" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) mock_exists = MagicMock(return_value=True) with patch.dict(zpool.__salt__, {'zpool.exists': mock_exists}): with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.scrub('mypool', pause=True) res = OrderedDict([('mypool', OrderedDict([('scrubbing', False)]))]) self.assertEqual(res, ret) def test_scrub_stop(self): ''' Tests pauze of scrub ''' ret = {} ret['stdout'] = "" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) mock_exists = MagicMock(return_value=True) with patch.dict(zpool.__salt__, {'zpool.exists': mock_exists}): with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.scrub('mypool', stop=True) res = OrderedDict([('mypool', OrderedDict([('scrubbing', False)]))]) self.assertEqual(res, ret) def test_split_success(self): ''' Tests split on success ''' ret = {} ret['stdout'] = "" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) mock_exists = Mock() mock_exists.side_effect = [False, True] with patch.dict(zpool.__salt__, {'zpool.exists': mock_exists}): with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.split('datapool', 'backuppool') res = OrderedDict([('backuppool', 'split off from datapool')]) self.assertEqual(res, ret) def test_split_exist_new(self): ''' Tests split on exising new pool ''' ret = {} ret['stdout'] = "" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) mock_exists = Mock() mock_exists.side_effect = [True, True] with patch.dict(zpool.__salt__, {'zpool.exists': mock_exists}): with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.split('datapool', 'backuppool') res = OrderedDict([('backuppool', 'storage pool already exists')]) self.assertEqual(res, ret) def test_split_missing_pool(self): ''' Tests split on missing source pool ''' ret = {} ret['stdout'] = "" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) mock_exists = Mock() mock_exists.side_effect = [False, False] with patch.dict(zpool.__salt__, {'zpool.exists': mock_exists}): with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.split('datapool', 'backuppool') res = OrderedDict([('datapool', 'storage pool does not exists')]) self.assertEqual(res, ret) def test_split_not_mirror(self): ''' Tests split on source pool is not a mirror ''' ret = {} ret['stdout'] = "" ret['stderr'] = "Unable to split datapool: Source pool must be composed only of mirrors" ret['retcode'] = 1 mock_cmd = MagicMock(return_value=ret) mock_exists = Mock() mock_exists.side_effect = [False, True] with patch.dict(zpool.__salt__, {'zpool.exists': mock_exists}): with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.split('datapool', 'backuppool') res = OrderedDict([('backuppool', 'Unable to split datapool: Source pool must be composed only of mirrors')]) self.assertEqual(res, ret) def test_labelclear_success(self): ''' Tests labelclear on succesful label removal ''' ret = {} ret['stdout'] = "" ret['stderr'] = "" ret['retcode'] = 0 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.labelclear('/dev/rdsk/c0t0d0', force=False) res = OrderedDict([('/dev/rdsk/c0t0d0', 'cleared')]) self.assertEqual(res, ret) def test_labelclear_cleared(self): ''' Tests labelclear on device with no label ''' ret = {} ret['stdout'] = "" ret['stderr'] = "failed to read label from /dev/rdsk/c0t0d0" ret['retcode'] = 1 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.labelclear('/dev/rdsk/c0t0d0', force=False) res = OrderedDict([('/dev/rdsk/c0t0d0', 'failed to read label from /dev/rdsk/c0t0d0')]) self.assertEqual(res, ret) def test_labelclear_exported(self): ''' Tests labelclear on device with from exported pool ''' ret = {} ret['stdout'] = "" ret['stderr'] = "\n".join([ "use '-f' to override the following error:", '/dev/rdsk/c0t0d0 is a member of exported pool "mypool"', ]) ret['retcode'] = 1 mock_cmd = MagicMock(return_value=ret) with patch.dict(zpool.__salt__, {'cmd.run_all': mock_cmd}): ret = zpool.labelclear('/dev/rdsk/c0t0d0', force=False) res = OrderedDict([('/dev/rdsk/c0t0d0', '/dev/rdsk/c0t0d0 is a member of exported pool "mypool"')]) self.assertEqual(res, ret)