self._FAKE_VOLUME_PATH)
drv._delete.assert_any_call(fake_vol_info)
- def _test_setup(self, config, share_config_exists=True):
- fake_exists = mock.Mock(return_value=share_config_exists)
+ @mock.patch('os.path.exists')
+ @mock.patch.object(image_utils, 'check_qemu_img_version')
+ def _test_setup(self, mock_check_qemu_img_version,
+ mock_exists, config, share_config_exists=True):
+ mock_exists.return_value = share_config_exists
fake_ensure_mounted = mock.MagicMock()
self._smbfs_driver._ensure_shares_mounted = fake_ensure_mounted
self._smbfs_driver.configuration = config
- with mock.patch('os.path.exists', fake_exists):
- if not (config.smbfs_shares_config and share_config_exists and
- config.smbfs_oversub_ratio > 0 and
- 0 <= config.smbfs_used_ratio <= 1):
- self.assertRaises(exception.SmbfsException,
- self._smbfs_driver.do_setup,
- None)
- else:
- self._smbfs_driver.do_setup(None)
- self.assertEqual(self._smbfs_driver.shares, {})
- fake_ensure_mounted.assert_called_once_with()
+ if not (config.smbfs_shares_config and share_config_exists and
+ config.smbfs_oversub_ratio > 0 and
+ 0 <= config.smbfs_used_ratio <= 1):
+ self.assertRaises(exception.SmbfsException,
+ self._smbfs_driver.do_setup,
+ None)
+ else:
+ self._smbfs_driver.do_setup(mock.sentinel.context)
+ mock_check_qemu_img_version.assert_called_once_with()
+ self.assertEqual(self._smbfs_driver.shares, {})
+ fake_ensure_mounted.assert_called_once_with()
def test_setup_missing_shares_config_option(self):
fake_config = copy.copy(self._FAKE_SMBFS_CONFIG)
fake_config.smbfs_shares_config = None
- self._test_setup(fake_config, None)
+ self._test_setup(config=fake_config,
+ share_config_exists=False)
def test_setup_missing_shares_config_file(self):
- self._test_setup(self._FAKE_SMBFS_CONFIG, False)
+ self._test_setup(config=self._FAKE_SMBFS_CONFIG,
+ share_config_exists=False)
def test_setup_invlid_oversub_ratio(self):
fake_config = copy.copy(self._FAKE_SMBFS_CONFIG)
fake_config.smbfs_oversub_ratio = -1
- self._test_setup(fake_config)
+ self._test_setup(config=fake_config)
def test_setup_invalid_used_ratio(self):
fake_config = copy.copy(self._FAKE_SMBFS_CONFIG)
fake_config.smbfs_used_ratio = -1
- self._test_setup(fake_config)
+ self._test_setup(config=fake_config)
def _test_create_volume(self, volume_exists=False, volume_format=None):
fake_method = mock.MagicMock()
self._smbfs_driver._remotefsclient.mount.assert_called_once_with(
self._FAKE_SHARE, self._FAKE_SHARE_OPTS.split())
- def _test_copy_image_to_volume(self, unsupported_qemu_version=False,
- wrong_size_after_fetch=False):
+ def _test_copy_image_to_volume(self, wrong_size_after_fetch=False):
drv = self._smbfs_driver
vol_size_bytes = self._FAKE_VOLUME['size'] << 30
- fake_image_service = mock.MagicMock()
- fake_image_service.show.return_value = (
- {'id': 'fake_image_id', 'disk_format': 'raw'})
fake_img_info = mock.MagicMock()
else:
fake_img_info.virtual_size = vol_size_bytes
- if unsupported_qemu_version:
- qemu_version = [1, 5]
- else:
- qemu_version = [1, 7]
-
drv.get_volume_format = mock.Mock(
return_value=drv._DISK_FORMAT_VHDX)
drv.local_path = mock.Mock(
return_value=self._FAKE_VOLUME_PATH)
- drv.get_qemu_version = mock.Mock(
- return_value=qemu_version)
drv._do_extend_volume = mock.Mock()
drv.configuration = mock.MagicMock()
drv.configuration.volume_dd_blocksize = (
mock.sentinel.block_size)
- exc = None
with mock.patch.object(image_utils, 'fetch_to_volume_format') as \
fake_fetch, mock.patch.object(image_utils, 'qemu_img_info') as \
fake_qemu_img_info:
- if wrong_size_after_fetch:
- exc = exception.ImageUnacceptable
- elif unsupported_qemu_version:
- exc = exception.InvalidVolume
-
fake_qemu_img_info.return_value = fake_img_info
- if exc:
+ if wrong_size_after_fetch:
self.assertRaises(
- exc, drv.copy_image_to_volume,
+ exception.ImageUnacceptable,
+ drv.copy_image_to_volume,
mock.sentinel.context, self._FAKE_VOLUME,
- fake_image_service,
+ mock.sentinel.image_service,
mock.sentinel.image_id)
else:
drv.copy_image_to_volume(
mock.sentinel.context, self._FAKE_VOLUME,
- fake_image_service,
+ mock.sentinel.image_service,
mock.sentinel.image_id)
fake_fetch.assert_called_once_with(
- mock.sentinel.context, fake_image_service,
+ mock.sentinel.context, mock.sentinel.image_service,
mock.sentinel.image_id, self._FAKE_VOLUME_PATH,
drv._DISK_FORMAT_VHDX,
mock.sentinel.block_size)
def test_copy_image_to_volume_wrong_size_after_fetch(self):
self._test_copy_image_to_volume(wrong_size_after_fetch=True)
- def test_copy_image_to_volume_unsupported_qemu_version(self):
- self._test_copy_image_to_volume(unsupported_qemu_version=True)
-
def test_get_capacity_info(self):
fake_block_size = 4096.0
fake_total_blocks = 1024
def test_copy_vhd_volume_to_image(self):
self._test_copy_volume_to_image(volume_format='vhd')
- def _test_copy_image_to_volume(self, qemu_version=[1, 7]):
+ def test_copy_image_to_volume(self):
drv = self._smbfs_driver
- fake_image_id = 'fake_image_id'
- fake_image_service = mock.MagicMock()
- fake_image_service.show.return_value = (
- {'id': fake_image_id, 'disk_format': 'raw'})
-
drv.get_volume_format = mock.Mock(
- return_value='vhdx')
+ return_value=mock.sentinel.volume_format)
drv.local_path = mock.Mock(
return_value=self._FAKE_VOLUME_PATH)
- drv._local_volume_dir = mock.Mock(
- return_value=self._FAKE_MNT_POINT)
- drv.get_qemu_version = mock.Mock(
- return_value=qemu_version)
drv.configuration = mock.MagicMock()
drv.configuration.volume_dd_blocksize = mock.sentinel.block_size
'fetch_to_volume_format') as fake_fetch:
drv.copy_image_to_volume(
mock.sentinel.context, self._FAKE_VOLUME,
- fake_image_service,
- fake_image_id)
-
- if qemu_version < [1, 7]:
- fake_temp_image_name = '%s.temp_image.%s.vhd' % (
- self._FAKE_VOLUME['id'],
- fake_image_id)
- fetch_path = os.path.join(self._FAKE_MNT_POINT,
- fake_temp_image_name)
- fetch_format = 'vpc'
- drv.vhdutils.convert_vhd.assert_called_once_with(
- fetch_path, self._FAKE_VOLUME_PATH)
- drv._delete.assert_called_with(fetch_path)
-
- else:
- fetch_path = self._FAKE_VOLUME_PATH
- fetch_format = 'vhdx'
-
+ mock.sentinel.image_service,
+ mock.sentinel.image_id)
fake_fetch.assert_called_once_with(
- mock.sentinel.context, fake_image_service, fake_image_id,
- fetch_path, fetch_format, mock.sentinel.block_size)
-
- def test_copy_image_to_volume(self):
- self._test_copy_image_to_volume()
-
- def test_copy_image_to_volume_with_conversion(self):
- self._test_copy_image_to_volume([1, 5])
+ mock.sentinel.context,
+ mock.sentinel.image_service,
+ mock.sentinel.image_id,
+ self._FAKE_VOLUME_PATH, mock.sentinel.volume_format,
+ mock.sentinel.block_size)
def test_copy_volume_from_snapshot(self):
drv = self._smbfs_driver
# under the License.
import os
-import re
from oslo_concurrency import processutils as putils
from oslo_config import cfg
from cinder.volume.drivers import remotefs as remotefs_drv
-VERSION = '1.0.0'
+VERSION = '1.1.0'
LOG = logging.getLogger(__name__)
SHARE_FORMAT_REGEX = r'//.+/.+'
VERSION = VERSION
+ _MINIMUM_QEMU_IMG_VERSION = '1.7'
+
_DISK_FORMAT_VHD = 'vhd'
_DISK_FORMAT_VHD_LEGACY = 'vpc'
_DISK_FORMAT_VHDX = 'vhdx'
}
def do_setup(self, context):
+ image_utils.check_qemu_img_version(self._MINIMUM_QEMU_IMG_VERSION)
+
config = self.configuration.smbfs_shares_config
if not config:
msg = (_("SMBFS config file not set (smbfs_shares_config)."))
info_path = self._local_path_volume_info(volume)
self._delete(info_path)
- def get_qemu_version(self):
- info, _ = self._execute('qemu-img', check_exit_code=False)
- pattern = r"qemu-img version ([0-9\.]*)"
- version = re.match(pattern, info)
- if not version:
- LOG.warn(_LW("qemu-img is not installed."))
- return None
- return [int(x) for x in version.groups()[0].split('.')]
-
def _create_windows_image(self, volume_path, volume_size, volume_format):
"""Creates a VHD or VHDX file of a given size."""
# vhd is regarded as vpc by qemu
if volume_format == self._DISK_FORMAT_VHD:
volume_format = self._DISK_FORMAT_VHD_LEGACY
- else:
- qemu_version = self.get_qemu_version()
- if qemu_version < [1, 7]:
- err_msg = _("This version of qemu-img does not support vhdx "
- "images. Please upgrade to 1.7 or greater.")
- raise exception.SmbfsException(err_msg)
self._execute('qemu-img', 'create', '-f', volume_format,
volume_path, str(volume_size * units.Gi),
def copy_image_to_volume(self, context, volume, image_service, image_id):
"""Fetch the image from image_service and write it to the volume."""
volume_format = self.get_volume_format(volume, qemu_format=True)
- image_meta = image_service.show(context, image_id)
- qemu_version = self.get_qemu_version()
-
- if (qemu_version < [1, 7] and (
- volume_format == self._DISK_FORMAT_VHDX and
- image_meta['disk_format'] != volume_format)):
- err_msg = _("Unsupported volume format: vhdx. qemu-img 1.7 or "
- "higher is required in order to properly support this "
- "format.")
- raise exception.InvalidVolume(err_msg)
image_utils.fetch_to_volume_format(
context, image_service, image_id,
from cinder.volume.drivers.windows import remotefs
from cinder.volume.drivers.windows import vhdutils
-VERSION = '1.0.0'
+VERSION = '1.1.0'
LOG = logging.getLogger(__name__)
class WindowsSmbfsDriver(smbfs.SmbfsDriver):
VERSION = VERSION
+ _MINIMUM_QEMU_IMG_VERSION = '1.6'
def __init__(self, *args, **kwargs):
super(WindowsSmbfsDriver, self).__init__(*args, **kwargs)
def copy_image_to_volume(self, context, volume, image_service, image_id):
"""Fetch the image from image_service and write it to the volume."""
+ volume_path = self.local_path(volume)
volume_format = self.get_volume_format(volume, qemu_format=True)
- image_meta = image_service.show(context, image_id)
-
- fetch_format = volume_format
- fetch_path = self.local_path(volume)
- self._delete(fetch_path)
- qemu_version = self.get_qemu_version()
-
- needs_conversion = False
-
- if (qemu_version < [1, 7] and (
- volume_format == self._DISK_FORMAT_VHDX and
- image_meta['disk_format'] != self._DISK_FORMAT_VHDX)):
- needs_conversion = True
- fetch_format = 'vpc'
- temp_file_name = '%s.temp_image.%s.%s' % (
- volume['id'],
- image_meta['id'],
- self._DISK_FORMAT_VHD)
- fetch_path = os.path.join(self._local_volume_dir(volume),
- temp_file_name)
+ self._delete(volume_path)
image_utils.fetch_to_volume_format(
context, image_service, image_id,
- fetch_path, fetch_format,
+ volume_path, volume_format,
self.configuration.volume_dd_blocksize)
- if needs_conversion:
- self.vhdutils.convert_vhd(fetch_path, self.local_path(volume))
- self._delete(fetch_path)
-
- self.vhdutils.resize_vhd(self.local_path(volume),
+ self.vhdutils.resize_vhd(volume_path,
volume['size'] * units.Gi)
def _copy_volume_from_snapshot(self, snapshot, volume, volume_size):
Code Review / openstack-build / cinder-build.git / commitdiff