"""Tests For miscellaneous util methods used with volume."""
-import os
-import re
-
import mock
from oslo.concurrency import processutils
from oslo.config import cfg
-from oslo.utils import importutils
-from cinder import context
-from cinder import db
from cinder import exception
from cinder.openstack.common import log as logging
from cinder import test
-from cinder.tests import fake_notifier
from cinder import utils
from cinder.volume import utils as volume_utils
CONF = cfg.CONF
-class UsageInfoTestCase(test.TestCase):
-
- QUEUE_NAME = 'cinder-volume'
- HOSTNAME = 'my-host.com'
- HOSTIP = '10.0.0.1'
- BACKEND = 'test_backend'
- MULTI_AT_BACKEND = 'test_b@ckend'
-
- def setUp(self):
- super(UsageInfoTestCase, self).setUp()
- self.addCleanup(fake_notifier.reset)
- self.flags(host='fake', notification_driver=["test"])
- self.volume = importutils.import_object(CONF.volume_manager)
- self.user_id = 'fake'
- self.project_id = 'fake'
- self.snapshot_id = 'fake'
- self.volume_size = 0
- self.context = context.RequestContext(self.user_id, self.project_id)
-
- def _create_volume(self, params=None):
- """Create a test volume."""
- params = params or {}
- vol = {}
- vol['snapshot_id'] = self.snapshot_id
- vol['user_id'] = self.user_id
- vol['project_id'] = self.project_id
- vol['host'] = CONF.host
- vol['availability_zone'] = CONF.storage_availability_zone
- vol['status'] = "creating"
- vol['attach_status'] = "detached"
- vol['size'] = self.volume_size
- vol.update(params)
- return db.volume_create(self.context, vol)['id']
+class NotifyUsageTestCase(test.TestCase):
+ @mock.patch('cinder.volume.utils._usage_from_volume')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_volume_usage(self, mock_rpc, mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_volume_usage(mock.sentinel.context,
+ mock.sentinel.volume,
+ 'test_suffix')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.volume)
+ mock_rpc.get_notifier.assert_called_once_with('volume', 'host1')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'volume.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_volume')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_volume_usage_with_kwargs(self, mock_rpc, mock_conf,
+ mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_volume_usage(
+ mock.sentinel.context,
+ mock.sentinel.volume,
+ 'test_suffix',
+ extra_usage_info={'a': 'b', 'c': 'd'},
+ host='host2')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.volume, a='b', c='d')
+ mock_rpc.get_notifier.assert_called_once_with('volume', 'host2')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'volume.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_volume')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_replication_usage(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_replication_usage(
+ mock.sentinel.context,
+ mock.sentinel.volume,
+ 'test_suffix')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.volume)
+ mock_rpc.get_notifier.assert_called_once_with('replication', 'host1')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'replication.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_volume')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_replication_usage_with_kwargs(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_replication_usage(
+ mock.sentinel.context,
+ mock.sentinel.volume,
+ 'test_suffix',
+ extra_usage_info={'a': 'b', 'c': 'd'},
+ host='host2')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.volume,
+ a='b', c='d')
+ mock_rpc.get_notifier.assert_called_once_with('replication', 'host2')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'replication.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_volume')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_replication_error(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_replication_error(
+ mock.sentinel.context,
+ mock.sentinel.volume,
+ 'test_suffix')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.volume)
+ mock_rpc.get_notifier.assert_called_once_with('replication', 'host1')
+ mock_rpc.get_notifier.return_value.error.assert_called_once_with(
+ mock.sentinel.context,
+ 'replication.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_volume')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_replication_error_with_kwargs(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_replication_error(
+ mock.sentinel.context,
+ mock.sentinel.volume,
+ 'test_suffix',
+ extra_error_info={'a': 'b', 'c': 'd'},
+ host='host2')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.volume,
+ a='b', c='d')
+ mock_rpc.get_notifier.assert_called_once_with('replication', 'host2')
+ mock_rpc.get_notifier.return_value.error.assert_called_once_with(
+ mock.sentinel.context,
+ 'replication.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_snapshot')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_snapshot_usage(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_snapshot_usage(
+ mock.sentinel.context,
+ mock.sentinel.snapshot,
+ 'test_suffix')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.snapshot)
+ mock_rpc.get_notifier.assert_called_once_with('snapshot', 'host1')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'snapshot.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_snapshot')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_snapshot_usage_with_kwargs(self, mock_rpc, mock_conf,
+ mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_snapshot_usage(
+ mock.sentinel.context,
+ mock.sentinel.snapshot,
+ 'test_suffix',
+ extra_usage_info={'a': 'b', 'c': 'd'},
+ host='host2')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.snapshot,
+ a='b', c='d')
+ mock_rpc.get_notifier.assert_called_once_with('snapshot', 'host2')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'snapshot.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_consistencygroup')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_consistencygroup_usage(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_consistencygroup_usage(
+ mock.sentinel.context,
+ mock.sentinel.consistencygroup,
+ 'test_suffix')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.consistencygroup)
+ mock_rpc.get_notifier.assert_called_once_with('consistencygroup',
+ 'host1')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'consistencygroup.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_consistencygroup')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_consistencygroup_usage_with_kwargs(self, mock_rpc,
+ mock_conf,
+ mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_consistencygroup_usage(
+ mock.sentinel.context,
+ mock.sentinel.consistencygroup,
+ 'test_suffix',
+ extra_usage_info={'a': 'b', 'c': 'd'},
+ host='host2')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.consistencygroup,
+ a='b', c='d')
+ mock_rpc.get_notifier.assert_called_once_with('consistencygroup',
+ 'host2')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'consistencygroup.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_cgsnapshot')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_cgsnapshot_usage(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_cgsnapshot_usage(
+ mock.sentinel.context,
+ mock.sentinel.cgsnapshot,
+ 'test_suffix')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.cgsnapshot)
+ mock_rpc.get_notifier.assert_called_once_with('cgsnapshot', 'host1')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'cgsnapshot.test_suffix',
+ mock_usage.return_value)
+
+ @mock.patch('cinder.volume.utils._usage_from_cgsnapshot')
+ @mock.patch('cinder.volume.utils.CONF')
+ @mock.patch('cinder.volume.utils.rpc')
+ def test_notify_about_cgsnapshot_usage_with_kwargs(self, mock_rpc,
+ mock_conf, mock_usage):
+ mock_conf.host = 'host1'
+ output = volume_utils.notify_about_cgsnapshot_usage(
+ mock.sentinel.context,
+ mock.sentinel.cgsnapshot,
+ 'test_suffix',
+ extra_usage_info={'a': 'b', 'c': 'd'},
+ host='host2')
+ self.assertIsNone(output)
+ mock_usage.assert_called_once_with(mock.sentinel.context,
+ mock.sentinel.cgsnapshot,
+ a='b', c='d')
+ mock_rpc.get_notifier.assert_called_once_with('cgsnapshot', 'host2')
+ mock_rpc.get_notifier.return_value.info.assert_called_once_with(
+ mock.sentinel.context,
+ 'cgsnapshot.test_suffix',
+ mock_usage.return_value)
class LVMVolumeDriverTestCase(test.TestCase):
self.assertEqual(count, 1024)
-class ClearVolumeTestCase(test.TestCase):
+class OdirectSupportTestCase(test.TestCase):
+ @mock.patch('cinder.utils.execute')
+ def test_check_for_odirect_support(self, mock_exec):
+ output = volume_utils.check_for_odirect_support('/dev/abc', '/dev/def')
+ self.assertTrue(output)
+ mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/abc',
+ 'of=/dev/def', 'oflag=direct',
+ run_as_root=True)
+ mock_exec.reset_mock()
+
+ output = volume_utils.check_for_odirect_support('/dev/abc', '/dev/def',
+ 'iflag=direct')
+ self.assertTrue(output)
+ mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/abc',
+ 'of=/dev/def', 'iflag=direct',
+ run_as_root=True)
+
+ @mock.patch('cinder.utils.execute',
+ side_effect=processutils.ProcessExecutionError)
+ def test_check_for_odirect_support_error(self, mock_exec):
+ output = volume_utils.check_for_odirect_support('/dev/abc', '/dev/def')
+ self.assertFalse(output)
+ mock_exec.assert_called_once_with('dd', 'count=0', 'if=/dev/abc',
+ 'of=/dev/def', 'oflag=direct',
+ run_as_root=True)
- def test_clear_volume(self):
- CONF.volume_clear = 'zero'
- CONF.volume_clear_size = 0
- CONF.volume_dd_blocksize = '1M'
- CONF.volume_clear_ionice = None
- self.mox.StubOutWithMock(volume_utils, 'copy_volume')
- volume_utils.copy_volume("/dev/zero", "volume_path", 1024,
- CONF.volume_dd_blocksize, sync=True,
- ionice=None, execute=utils.execute)
- self.mox.ReplayAll()
- volume_utils.clear_volume(1024, "volume_path")
-
- def test_clear_volume_zero(self):
- CONF.volume_clear = 'zero'
- CONF.volume_clear_size = 1
- CONF.volume_clear_ionice = None
- self.mox.StubOutWithMock(volume_utils, 'copy_volume')
- volume_utils.copy_volume("/dev/zero", "volume_path", 1,
- CONF.volume_dd_blocksize, sync=True,
- ionice=None, execute=utils.execute)
- self.mox.ReplayAll()
- volume_utils.clear_volume(1024, "volume_path")
-
- def test_clear_volume_ionice(self):
- CONF.volume_clear = 'zero'
- CONF.volume_clear_size = 0
- CONF.volume_dd_blocksize = '1M'
- CONF.volume_clear_ionice = '-c3'
- self.mox.StubOutWithMock(volume_utils, 'copy_volume')
- volume_utils.copy_volume("/dev/zero", "volume_path", 1024,
- CONF.volume_dd_blocksize, sync=True,
- ionice=CONF.volume_clear_ionice,
- execute=utils.execute)
- self.mox.ReplayAll()
- volume_utils.clear_volume(1024, "volume_path")
-
- def test_clear_volume_zero_ionice(self):
- CONF.volume_clear = 'zero'
- CONF.volume_clear_size = 1
- CONF.volume_clear_ionice = '-c3'
- self.mox.StubOutWithMock(volume_utils, 'copy_volume')
- volume_utils.copy_volume("/dev/zero", "volume_path", 1,
- CONF.volume_dd_blocksize, sync=True,
- ionice=CONF.volume_clear_ionice,
- execute=utils.execute)
- self.mox.ReplayAll()
- volume_utils.clear_volume(1024, "volume_path")
-
- def test_clear_volume_shred(self):
- CONF.volume_clear = 'shred'
- CONF.volume_clear_size = 1
- clear_cmd = ['shred', '-n3', '-s1MiB', "volume_path"]
- self.mox.StubOutWithMock(utils, "execute")
- utils.execute(*clear_cmd, run_as_root=True)
- self.mox.ReplayAll()
- volume_utils.clear_volume(1024, "volume_path")
-
- def test_clear_volume_shred_not_clear_size(self):
- CONF.volume_clear = 'shred'
- CONF.volume_clear_size = None
- clear_cmd = ['shred', '-n3', "volume_path"]
- self.mox.StubOutWithMock(utils, "execute")
- utils.execute(*clear_cmd, run_as_root=True)
- self.mox.ReplayAll()
- volume_utils.clear_volume(1024, "volume_path")
-
- def test_clear_volume_invalid_opt(self):
- CONF.volume_clear = 'non_existent_volume_clearer'
- CONF.volume_clear_size = 0
- self.mox.StubOutWithMock(volume_utils, 'copy_volume')
-
- self.mox.ReplayAll()
+class ClearVolumeTestCase(test.TestCase):
+ @mock.patch('cinder.volume.utils.copy_volume', return_value=None)
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_clear_volume_conf(self, mock_conf, mock_copy):
+ mock_conf.volume_clear = 'zero'
+ mock_conf.volume_clear_size = 0
+ mock_conf.volume_dd_blocksize = '1M'
+ mock_conf.volume_clear_ionice = '-c3'
+ output = volume_utils.clear_volume(1024, 'volume_path')
+ self.assertIsNone(output)
+ mock_copy.assert_called_once_with('/dev/zero', 'volume_path', 1024,
+ '1M', sync=True,
+ execute=utils.execute, ionice='-c3')
+
+ @mock.patch('cinder.volume.utils.copy_volume', return_value=None)
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_clear_volume_args(self, mock_conf, mock_copy):
+ mock_conf.volume_clear = 'shred'
+ mock_conf.volume_clear_size = 0
+ mock_conf.volume_dd_blocksize = '1M'
+ mock_conf.volume_clear_ionice = '-c3'
+ output = volume_utils.clear_volume(1024, 'volume_path', 'zero', 1,
+ '-c0')
+ self.assertIsNone(output)
+ mock_copy.assert_called_once_with('/dev/zero', 'volume_path', 1,
+ '1M', sync=True,
+ execute=utils.execute, ionice='-c0')
+
+ @mock.patch('cinder.utils.execute')
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_clear_volume_shred(self, mock_conf, mock_exec):
+ mock_conf.volume_clear = 'shred'
+ mock_conf.volume_clear_size = 1
+ mock_conf.volume_clear_ionice = None
+ output = volume_utils.clear_volume(1024, 'volume_path')
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with(
+ 'shred', '-n3', '-s1MiB', "volume_path", run_as_root=True)
+
+ @mock.patch('cinder.utils.execute')
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_clear_volume_shred_not_clear_size(self, mock_conf, mock_exec):
+ mock_conf.volume_clear = 'shred'
+ mock_conf.volume_clear_size = None
+ mock_conf.volume_clear_ionice = None
+ output = volume_utils.clear_volume(1024, 'volume_path')
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with(
+ 'shred', '-n3', "volume_path", run_as_root=True)
+
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_clear_volume_invalid_opt(self, mock_conf):
+ mock_conf.volume_clear = 'non_existent_volume_clearer'
+ mock_conf.volume_clear_size = 0
+ mock_conf.volume_clear_ionice = None
self.assertRaises(exception.InvalidConfigurationValue,
volume_utils.clear_volume,
1024, "volume_path")
- def test_clear_volume_lvm_snap(self):
- self.stubs.Set(os.path, 'exists', lambda x: True)
- CONF.volume_clear = 'zero'
- CONF.volume_clear_size = 0
-
- uuid = '00000000-0000-0000-0000-90ed32cdeed3'
- name = 'snapshot-' + uuid
- mangle_name = '_' + re.sub(r'-', r'--', name)
- vol_path = '/dev/mapper/cinder--volumes-%s-cow' % mangle_name
-
- def fake_copy_volume(srcstr, deststr, size, blocksize, **kwargs):
- self.assertEqual(deststr, vol_path)
- return True
-
- self.stubs.Set(volume_utils, 'copy_volume', fake_copy_volume)
- volume_utils.clear_volume(123, vol_path)
-
class CopyVolumeTestCase(test.TestCase):
-
- def test_copy_volume_dd_iflag_and_oflag(self):
- def fake_utils_execute(*cmd, **kwargs):
- if 'if=/dev/zero' in cmd and 'iflag=direct' in cmd:
- raise processutils.ProcessExecutionError()
- if 'of=/dev/null' in cmd and 'oflag=direct' in cmd:
- raise processutils.ProcessExecutionError()
- if 'iflag=direct' in cmd and 'oflag=direct' in cmd:
- raise exception.InvalidInput(message='iflag/oflag error')
-
- def fake_check_odirect(src, dest, flags='blah'):
- return False
-
- self.stubs.Set(volume_utils,
- 'check_for_odirect_support',
- fake_check_odirect)
-
- volume_utils.copy_volume('/dev/zero', '/dev/null', 1024,
- CONF.volume_dd_blocksize, sync=True,
- ionice=None, execute=fake_utils_execute)
+ @mock.patch('cinder.volume.utils.setup_blkio_cgroup',
+ return_value=['cg_cmd'])
+ @mock.patch('cinder.volume.utils._calculate_count',
+ return_value=(1234, 5678))
+ @mock.patch('cinder.volume.utils.check_for_odirect_support',
+ return_value=True)
+ @mock.patch('cinder.utils.execute')
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_copy_volume_dd_iflag_and_oflag(self, mock_conf, mock_exec,
+ mock_support, mock_count, mock_cg):
+ mock_conf.volume_copy_bps_limit = 10
+ output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, 1,
+ sync=True, execute=utils.execute,
+ ionice=None)
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with('cg_cmd', 'dd', 'if=/dev/zero',
+ 'of=/dev/null', 'count=5678',
+ 'bs=1234', 'iflag=direct',
+ 'oflag=direct', run_as_root=True)
+
+ mock_exec.reset_mock()
+
+ output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, 1,
+ sync=False, execute=utils.execute,
+ ionice=None)
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with('cg_cmd', 'dd', 'if=/dev/zero',
+ 'of=/dev/null', 'count=5678',
+ 'bs=1234', 'iflag=direct',
+ 'oflag=direct', run_as_root=True)
+
+ @mock.patch('cinder.volume.utils.setup_blkio_cgroup',
+ return_value=['cg_cmd'])
+ @mock.patch('cinder.volume.utils._calculate_count',
+ return_value=(1234, 5678))
+ @mock.patch('cinder.volume.utils.check_for_odirect_support',
+ return_value=False)
+ @mock.patch('cinder.utils.execute')
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_copy_volume_dd_no_iflag_or_oflag(self, mock_conf, mock_exec,
+ mock_support, mock_count,
+ mock_cg):
+ mock_conf.volume_copy_bps_limit = 10
+ output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, 1,
+ sync=True, execute=utils.execute,
+ ionice=None)
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with('cg_cmd', 'dd', 'if=/dev/zero',
+ 'of=/dev/null', 'count=5678',
+ 'bs=1234', 'conv=fdatasync',
+ run_as_root=True)
+
+ mock_exec.reset_mock()
+
+ output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, 1,
+ sync=False, execute=utils.execute,
+ ionice=None)
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with('cg_cmd', 'dd', 'if=/dev/zero',
+ 'of=/dev/null', 'count=5678',
+ 'bs=1234', run_as_root=True)
+
+ @mock.patch('cinder.volume.utils.setup_blkio_cgroup',
+ return_value=None)
+ @mock.patch('cinder.volume.utils._calculate_count',
+ return_value=(1234, 5678))
+ @mock.patch('cinder.volume.utils.check_for_odirect_support',
+ return_value=False)
+ @mock.patch('cinder.utils.execute')
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_copy_volume_dd_no_cgroup(self, mock_conf, mock_exec, mock_support,
+ mock_count, mock_cg):
+ mock_conf.volume_copy_bps_limit = 10
+ output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, 1,
+ sync=True, execute=utils.execute,
+ ionice=None)
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with('dd', 'if=/dev/zero', 'of=/dev/null',
+ 'count=5678', 'bs=1234',
+ 'conv=fdatasync', run_as_root=True)
+
+ @mock.patch('cinder.volume.utils.setup_blkio_cgroup',
+ return_value=None)
+ @mock.patch('cinder.volume.utils._calculate_count',
+ return_value=(1234, 5678))
+ @mock.patch('cinder.volume.utils.check_for_odirect_support',
+ return_value=False)
+ @mock.patch('cinder.utils.execute')
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_copy_volume_dd_with_ionice(self, mock_conf, mock_exec,
+ mock_support, mock_count, mock_cg):
+ mock_conf.volume_copy_bps_limit = 10
+ output = volume_utils.copy_volume('/dev/zero', '/dev/null', 1024, 1,
+ sync=True, execute=utils.execute,
+ ionice='-c3')
+ self.assertIsNone(output)
+ mock_exec.assert_called_once_with('ionice', '-c3', 'dd',
+ 'if=/dev/zero', 'of=/dev/null',
+ 'count=5678', 'bs=1234',
+ 'conv=fdatasync', run_as_root=True)
class BlkioCgroupTestCase(test.TestCase):
-
- @mock.patch.object(utils, 'get_blkdev_major_minor')
- def test_setup_blkio_cgroup(self, mock_major_minor):
-
- def fake_get_blkdev_major_minor(path):
- return {'src_volume': "253:0", 'dst_volume': "253:1"}[path]
-
- mock_major_minor.side_effect = fake_get_blkdev_major_minor
-
- self.exec_cnt = 0
-
- def fake_utils_execute(*cmd, **kwargs):
- exec_cmds = [('cgcreate', '-g',
- 'blkio:' + CONF.volume_copy_blkio_cgroup_name),
- ('cgset', '-r',
- 'blkio.throttle.read_bps_device=253:0 1024',
- CONF.volume_copy_blkio_cgroup_name),
- ('cgset', '-r',
- 'blkio.throttle.write_bps_device=253:1 1024',
- CONF.volume_copy_blkio_cgroup_name)]
- self.assertEqual(exec_cmds[self.exec_cnt], cmd)
- self.exec_cnt += 1
-
- cmd = volume_utils.setup_blkio_cgroup('src_volume', 'dst_volume', 1024,
- execute=fake_utils_execute)
- self.assertEqual(['cgexec', '-g',
- 'blkio:' + CONF.volume_copy_blkio_cgroup_name], cmd)
+ def test_bps_limit_zero(self):
+ mock_exec = mock.Mock()
+ output = volume_utils.setup_blkio_cgroup('src', 'dst', 0,
+ execute=mock_exec)
+ self.assertIsNone(output)
+ self.assertFalse(mock_exec.called)
+
+ @mock.patch('cinder.utils.get_blkdev_major_minor',
+ side_effect=exception.Error)
+ def test_get_blkdev_error(self, mock_get_blkdev):
+ mock_exec = mock.Mock()
+ output = volume_utils.setup_blkio_cgroup('src', 'dst', 1,
+ execute=mock_exec)
+ self.assertIsNone(output)
+ mock_get_blkdev.assert_has_calls([mock.call('src'), mock.call('dst')])
+ self.assertFalse(mock_exec.called)
+
+ @mock.patch('cinder.utils.get_blkdev_major_minor',
+ side_effect=lambda x: x)
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_cgcreate_fail(self, mock_conf, mock_get_blkdev):
+ mock_conf.volume_copy_blkio_cgroup_name = 'test_group'
+ mock_exec = mock.Mock()
+ mock_exec.side_effect = processutils.ProcessExecutionError
+ output = volume_utils.setup_blkio_cgroup('src', 'dst', 1,
+ execute=mock_exec)
+ self.assertIsNone(output)
+ mock_get_blkdev.assert_has_calls([mock.call('src'), mock.call('dst')])
+ mock_exec.assert_called_once_with('cgcreate', '-g', 'blkio:test_group',
+ run_as_root=True)
+
+ @mock.patch('cinder.utils.get_blkdev_major_minor',
+ side_effect=lambda x: x)
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_cgset_fail(self, mock_conf, mock_get_blkdev):
+ mock_conf.volume_copy_blkio_cgroup_name = 'test_group'
+ mock_exec = mock.Mock()
+
+ def cgset_exception(*args, **kwargs):
+ if 'cgset' in args:
+ raise processutils.ProcessExecutionError
+
+ mock_exec.side_effect = cgset_exception
+ output = volume_utils.setup_blkio_cgroup('src', 'dst', 1,
+ execute=mock_exec)
+ self.assertIsNone(output)
+ mock_get_blkdev.assert_has_calls([mock.call('src'), mock.call('dst')])
+ mock_exec.assert_has_calls([
+ mock.call('cgcreate', '-g', 'blkio:test_group', run_as_root=True),
+ mock.call('cgset', '-r', 'blkio.throttle.read_bps_device=src 1',
+ 'test_group', run_as_root=True)])
+
+ @mock.patch('cinder.utils.get_blkdev_major_minor',
+ side_effect=lambda x: x)
+ @mock.patch('cinder.volume.utils.CONF')
+ def test_setup_blkio_cgroup(self, mock_conf, mock_get_blkdev):
+ mock_conf.volume_copy_blkio_cgroup_name = 'test_group'
+ mock_exec = mock.Mock()
+ output = volume_utils.setup_blkio_cgroup('src', 'dst', 1,
+ execute=mock_exec)
+ self.assertEqual(['cgexec', '-g', 'blkio:test_group'], output)
+ mock_get_blkdev.assert_has_calls([mock.call('src'), mock.call('dst')])
+ mock_exec.assert_has_calls([
+ mock.call('cgcreate', '-g', 'blkio:test_group', run_as_root=True),
+ mock.call('cgset', '-r', 'blkio.throttle.read_bps_device=src 1',
+ 'test_group', run_as_root=True),
+ mock.call('cgset', '-r', 'blkio.throttle.write_bps_device=dst 1',
+ 'test_group', run_as_root=True)])
class VolumeUtilsTestCase(test.TestCase):
+ def test_null_safe_str(self):
+ self.assertEqual('', volume_utils.null_safe_str(None))
+ self.assertEqual('', volume_utils.null_safe_str(False))
+ self.assertEqual('', volume_utils.null_safe_str(0))
+ self.assertEqual('', volume_utils.null_safe_str([]))
+ self.assertEqual('', volume_utils.null_safe_str(()))
+ self.assertEqual('', volume_utils.null_safe_str({}))
+ self.assertEqual('', volume_utils.null_safe_str(set()))
+ self.assertEqual('a', volume_utils.null_safe_str('a'))
+ self.assertEqual('1', volume_utils.null_safe_str(1))
+ self.assertEqual('True', volume_utils.null_safe_str(True))
+
+ @mock.patch('cinder.utils.get_root_helper')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning')
+ def test_supports_thin_provisioning(self, mock_supports_thin, mock_helper):
+ self.assertEqual(mock_supports_thin.return_value,
+ volume_utils.supports_thin_provisioning())
+ mock_helper.assert_called_once_with()
+
+ @mock.patch('cinder.utils.get_root_helper')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
+ def test_get_all_physical_volumes(self, mock_get_vols, mock_helper):
+ self.assertEqual(mock_get_vols.return_value,
+ volume_utils.get_all_physical_volumes())
+ mock_helper.assert_called_once_with()
+
+ @mock.patch('cinder.utils.get_root_helper')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_volume_groups')
+ def test_get_all_volume_groups(self, mock_get_groups, mock_helper):
+ self.assertEqual(mock_get_groups.return_value,
+ volume_utils.get_all_volume_groups())
+ mock_helper.assert_called_once_with()
+
def test_generate_password(self):
password = volume_utils.generate_password()
- self.assertTrue([c for c in password if c in '0123456789'])
- self.assertTrue([c for c in password
- if c in 'abcdefghijklmnopqrstuvwxyz'])
- self.assertTrue([c for c in password
- if c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ'])
+ self.assertTrue(any(c for c in password if c in '23456789'))
+ self.assertTrue(any(c for c in password
+ if c in 'abcdefghijkmnopqrstuvwxyz'))
+ self.assertTrue(any(c for c in password
+ if c in 'ABCDEFGHJKLMNPQRSTUVWXYZ'))
+ self.assertEqual(16, len(password))
+ self.assertEqual(10, len(volume_utils.generate_password(10)))
+
+ @mock.patch('cinder.volume.utils.generate_password')
+ def test_generate_username(self, mock_gen_pass):
+ output = volume_utils.generate_username()
+ self.assertEqual(mock_gen_pass.return_value, output)
def test_extract_host(self):
host = 'Host'
Code Review / openstack-build / cinder-build.git / commitdiff