openstack/nova
Code Issues Proposed changes
nova/nova/tests/unit/virt/libvirt/test_utils.py
Mikhail Feoktistov d4aa455d53 libvirt: virtuozzo instance resize support 
Adapt "nova resize" code to support Virtuozzo ploop disks.
As far as ploop disks are in fact directories we add '-r' argument
to all utilities that deal with instance' disks such as cp, rsync and scp.
Thus we copy disks universally whether they are folders or files.

Also using "prl_disk_tool" instead of "qemu-img" is better for ploop images
because it resizes guest filesystem as well.

We can't resize disks from guest OS in containers,
because they are not allowed to write directly to block device.
ploop tool can resize partition table and internal filesystem,
but only for container's disks. Such disks must have only one partition
with ext filesystem.

prl_disk_tool can resize disks with internal filesystems
and doesn't require any special layout so it can resize disks
for virtual machines.  So it's better to use this tool instead of ploop.

Also we make compute.filters more strict
We call "ploop" only with "restore-descriptor" argument
And we set disk size in megabytes for prl_disk_tool

Co-Authored-By: Dmitry Guryanov <dguryanov@parallels.com>
Depends-On: I04c4379459c2fc1fd4801ec2aad53d0f6053b6d6
Change-Id: I38dbf73beb01fe1939ddca63fbfedbec1dc3c826
Implements: blueprint virtuozzo-instance-resize-support
2016-06-28 22:13:49 +03:00

739 lines
30 KiB
Python
Raw Blame History

# Copyright 2012 NTT Data. All Rights Reserved.
# Copyright 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import functools
import os
import tempfile
import mock
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import fileutils
import six
from nova.compute import arch
from nova import context
from nova import exception
from nova import objects
from nova import test
from nova.tests.unit import fake_instance
from nova.tests import uuidsentinel as uuids
from nova import utils
from nova.virt.disk import api as disk
from nova.virt import images
from nova.virt.libvirt import config as vconfig
from nova.virt.libvirt import utils as libvirt_utils
CONF = cfg.CONF
class LibvirtUtilsTestCase(test.NoDBTestCase):
@mock.patch('nova.utils.execute')
def test_copy_image_local(self, mock_execute):
libvirt_utils.copy_image('src', 'dest')
mock_execute.assert_called_once_with('cp', '-r', 'src', 'dest')
@mock.patch('nova.virt.libvirt.volume.remotefs.SshDriver.copy_file')
def test_copy_image_remote_ssh(self, mock_rem_fs_remove):
self.flags(remote_filesystem_transport='ssh', group='libvirt')
libvirt_utils.copy_image('src', 'dest', host='host')
mock_rem_fs_remove.assert_called_once_with('src', 'host:dest',
on_completion=None, on_execute=None, compression=True)
@mock.patch('nova.virt.libvirt.volume.remotefs.RsyncDriver.copy_file')
def test_copy_image_remote_rsync(self, mock_rem_fs_remove):
self.flags(remote_filesystem_transport='rsync', group='libvirt')
libvirt_utils.copy_image('src', 'dest', host='host')
mock_rem_fs_remove.assert_called_once_with('src', 'host:dest',
on_completion=None, on_execute=None, compression=True)
@mock.patch('os.path.exists', return_value=True)
def test_disk_type_from_path(self, mock_exists):
# Seems like lvm detection
# if its in /dev ??
for p in ['/dev/b', '/dev/blah/blah']:
d_type = libvirt_utils.get_disk_type_from_path(p)
self.assertEqual('lvm', d_type)
# Try rbd detection
d_type = libvirt_utils.get_disk_type_from_path('rbd:pool/instance')
self.assertEqual('rbd', d_type)
# Try the other types
path = '/myhome/disk.config'
d_type = libvirt_utils.get_disk_type_from_path(path)
self.assertIsNone(d_type)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('os.path.isdir', return_value=True)
def test_disk_type_ploop(self, mock_isdir, mock_exists):
path = '/some/path'
d_type = libvirt_utils.get_disk_type_from_path(path)
mock_isdir.assert_called_once_with(path)
mock_exists.assert_called_once_with("%s/DiskDescriptor.xml" % path)
self.assertEqual('ploop', d_type)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_disk_backing(self, mock_execute, mock_exists):
path = '/myhome/disk.config'
template_output = """image: %(path)s
file format: raw
virtual size: 2K (2048 bytes)
cluster_size: 65536
disk size: 96K
"""
output = template_output % ({
'path': path,
})
mock_execute.return_value = (output, '')
d_backing = libvirt_utils.get_disk_backing_file(path)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
mock_exists.assert_called_once_with(path)
self.assertIsNone(d_backing)
def _test_disk_size(self, mock_execute, path, expected_size):
d_size = libvirt_utils.get_disk_size(path)
self.assertEqual(expected_size, d_size)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
@mock.patch('os.path.exists', return_value=True)
def test_disk_size(self, mock_exists):
path = '/myhome/disk.config'
template_output = """image: %(path)s
file format: raw
virtual size: %(v_size)s (%(vsize_b)s bytes)
cluster_size: 65536
disk size: 96K
"""
for i in range(0, 128):
bytes = i * 65336
kbytes = bytes / 1024
mbytes = kbytes / 1024
output = template_output % ({
'v_size': "%sM" % (mbytes),
'vsize_b': i,
'path': path,
})
with mock.patch('nova.utils.execute',
return_value=(output, '')) as mock_execute:
self._test_disk_size(mock_execute, path, i)
output = template_output % ({
'v_size': "%sK" % (kbytes),
'vsize_b': i,
'path': path,
})
with mock.patch('nova.utils.execute',
return_value=(output, '')) as mock_execute:
self._test_disk_size(mock_execute, path, i)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_qemu_info_canon(self, mock_execute, mock_exists):
path = "disk.config"
example_output = """image: disk.config
file format: raw
virtual size: 64M (67108864 bytes)
cluster_size: 65536
disk size: 96K
blah BLAH: bb
"""
mock_execute.return_value = (example_output, '')
image_info = images.qemu_img_info(path)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
mock_exists.assert_called_once_with(path)
self.assertEqual('disk.config', image_info.image)
self.assertEqual('raw', image_info.file_format)
self.assertEqual(67108864, image_info.virtual_size)
self.assertEqual(98304, image_info.disk_size)
self.assertEqual(65536, image_info.cluster_size)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_qemu_info_canon2(self, mock_execute, mock_exists):
path = "disk.config"
example_output = """image: disk.config
file format: QCOW2
virtual size: 67108844
cluster_size: 65536
disk size: 963434
backing file: /var/lib/nova/a328c7998805951a_2
"""
mock_execute.return_value = (example_output, '')
image_info = images.qemu_img_info(path)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
mock_exists.assert_called_once_with(path)
self.assertEqual('disk.config', image_info.image)
self.assertEqual('qcow2', image_info.file_format)
self.assertEqual(67108844, image_info.virtual_size)
self.assertEqual(963434, image_info.disk_size)
self.assertEqual(65536, image_info.cluster_size)
self.assertEqual('/var/lib/nova/a328c7998805951a_2',
image_info.backing_file)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('os.path.isdir', return_value=True)
@mock.patch('nova.utils.execute')
def test_qemu_info_ploop(self, mock_execute, mock_isdir, mock_exists):
path = "/var/lib/nova"
example_output = """image: root.hds
file format: parallels
virtual size: 3.0G (3221225472 bytes)
disk size: 706M
"""
mock_execute.return_value = (example_output, '')
image_info = images.qemu_img_info(path)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info',
os.path.join(path, 'root.hds'),
prlimit=images.QEMU_IMG_LIMITS)
mock_isdir.assert_called_once_with(path)
self.assertEqual(2, mock_exists.call_count)
self.assertEqual(path, mock_exists.call_args_list[0][0][0])
self.assertEqual(os.path.join(path, 'DiskDescriptor.xml'),
mock_exists.call_args_list[1][0][0])
self.assertEqual('root.hds', image_info.image)
self.assertEqual('parallels', image_info.file_format)
self.assertEqual(3221225472, image_info.virtual_size)
self.assertEqual(740294656, image_info.disk_size)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_qemu_backing_file_actual(self,
mock_execute, mock_exists):
path = "disk.config"
example_output = """image: disk.config
file format: raw
virtual size: 64M (67108864 bytes)
cluster_size: 65536
disk size: 96K
Snapshot list:
ID TAG VM SIZE DATE VM CLOCK
1 d9a9784a500742a7bb95627bb3aace38 0 2012-08-20 10:52:46 00:00:00.000
backing file: /var/lib/nova/a328c7998805951a_2 (actual path: /b/3a988059e51a_2)
"""
mock_execute.return_value = (example_output, '')
image_info = images.qemu_img_info(path)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
mock_exists.assert_called_once_with(path)
self.assertEqual('disk.config', image_info.image)
self.assertEqual('raw', image_info.file_format)
self.assertEqual(67108864, image_info.virtual_size)
self.assertEqual(98304, image_info.disk_size)
self.assertEqual(1, len(image_info.snapshots))
self.assertEqual('/b/3a988059e51a_2',
image_info.backing_file)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_qemu_info_convert(self, mock_execute, mock_exists):
path = "disk.config"
example_output = """image: disk.config
file format: raw
virtual size: 64M
disk size: 96K
Snapshot list:
ID TAG VM SIZE DATE VM CLOCK
1 d9a9784a500742a7bb95627bb3aace38 0 2012-08-20 10:52:46 00:00:00.000
3 d9a9784a500742a7bb95627bb3aace38 0 2012-08-20 10:52:46 00:00:00.000
4 d9a9784a500742a7bb95627bb3aace38 0 2012-08-20 10:52:46 00:00:00.000
junk stuff: bbb
"""
mock_execute.return_value = (example_output, '')
image_info = images.qemu_img_info(path)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
mock_exists.assert_called_once_with(path)
self.assertEqual('disk.config', image_info.image)
self.assertEqual('raw', image_info.file_format)
self.assertEqual(67108864, image_info.virtual_size)
self.assertEqual(98304, image_info.disk_size)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_qemu_info_snaps(self, mock_execute, mock_exists):
path = "disk.config"
example_output = """image: disk.config
file format: raw
virtual size: 64M (67108864 bytes)
disk size: 96K
Snapshot list:
ID TAG VM SIZE DATE VM CLOCK
1 d9a9784a500742a7bb95627bb3aace38 0 2012-08-20 10:52:46 00:00:00.000
3 d9a9784a500742a7bb95627bb3aace38 0 2012-08-20 10:52:46 00:00:00.000
4 d9a9784a500742a7bb95627bb3aace38 0 2012-08-20 10:52:46 00:00:00.000
"""
mock_execute.return_value = (example_output, '')
image_info = images.qemu_img_info(path)
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
mock_exists.assert_called_once_with(path)
self.assertEqual('disk.config', image_info.image)
self.assertEqual('raw', image_info.file_format)
self.assertEqual(67108864, image_info.virtual_size)
self.assertEqual(98304, image_info.disk_size)
self.assertEqual(3, len(image_info.snapshots))
def test_valid_hostname_normal(self):
self.assertTrue(libvirt_utils.is_valid_hostname("hello.world.com"))
def test_valid_hostname_ipv4addr(self):
self.assertTrue(libvirt_utils.is_valid_hostname("10.0.2.1"))
def test_valid_hostname_ipv6addr(self):
self.assertTrue(libvirt_utils.is_valid_hostname("240:2ac3::2"))
def test_valid_hostname_bad(self):
self.assertFalse(libvirt_utils.is_valid_hostname("foo/?com=/bin/sh"))
@mock.patch('nova.utils.execute')
def test_create_image(self, mock_execute):
libvirt_utils.create_image('raw', '/some/path', '10G')
libvirt_utils.create_image('qcow2', '/some/stuff', '1234567891234')
expected_args = [(('qemu-img', 'create', '-f', 'raw',
'/some/path', '10G'),),
(('qemu-img', 'create', '-f', 'qcow2',
'/some/stuff', '1234567891234'),)]
self.assertEqual(expected_args, mock_execute.call_args_list)
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_create_cow_image(self, mock_execute, mock_exists):
mock_execute.return_value = ('stdout', None)
libvirt_utils.create_cow_image('/some/path', '/the/new/cow')
expected_args = [(('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', '/some/path'),
{'prlimit': images.QEMU_IMG_LIMITS}),
(('qemu-img', 'create', '-f', 'qcow2',
'-o', 'backing_file=/some/path',
'/the/new/cow'),)]
self.assertEqual(expected_args, mock_execute.call_args_list)
def test_pick_disk_driver_name(self):
type_map = {'kvm': ([True, 'qemu'], [False, 'qemu'], [None, 'qemu']),
'qemu': ([True, 'qemu'], [False, 'qemu'], [None, 'qemu']),
'uml': ([True, None], [False, None], [None, None]),
'lxc': ([True, None], [False, None], [None, None])}
# NOTE(aloga): Xen is tested in test_pick_disk_driver_name_xen
version = 1005001
for (virt_type, checks) in six.iteritems(type_map):
self.flags(virt_type=virt_type, group='libvirt')
for (is_block_dev, expected_result) in checks:
result = libvirt_utils.pick_disk_driver_name(version,
is_block_dev)
self.assertEqual(result, expected_result)
@mock.patch('nova.utils.execute')
def test_pick_disk_driver_name_xen(self, mock_execute):
def side_effect(*args, **kwargs):
if args == ('tap-ctl', 'check'):
if mock_execute.blktap is True:
return ('ok\n', '')
elif mock_execute.blktap is False:
return ('some error\n', '')
else:
raise OSError(2, "No such file or directory")
elif args == ('xend', 'status'):
if mock_execute.xend is True:
return ('', '')
elif mock_execute.xend is False:
raise processutils.ProcessExecutionError("error")
else:
raise OSError(2, "No such file or directory")
raise Exception('Unexpected call')
mock_execute.side_effect = side_effect
self.flags(virt_type="xen", group='libvirt')
versions = [4000000, 4001000, 4002000, 4003000, 4005000]
for version in versions:
# block dev
result = libvirt_utils.pick_disk_driver_name(version, True)
self.assertEqual(result, "phy")
self.assertFalse(mock_execute.called)
mock_execute.reset_mock()
# file dev
for blktap in True, False, None:
mock_execute.blktap = blktap
for xend in True, False, None:
mock_execute.xend = xend
result = libvirt_utils.pick_disk_driver_name(version,
False)
# qemu backend supported only by libxl which is
# production since xen 4.2. libvirt use libxl if
# xend service not started.
if version >= 4002000 and xend is not True:
self.assertEqual(result, 'qemu')
elif blktap:
if version == 4000000:
self.assertEqual(result, 'tap')
else:
self.assertEqual(result, 'tap2')
else:
self.assertEqual(result, 'file')
# default is_block_dev False
self.assertEqual(result,
libvirt_utils.pick_disk_driver_name(version))
mock_execute.reset_mock()
@mock.patch('os.path.exists', return_value=True)
@mock.patch('nova.utils.execute')
def test_get_disk_size(self, mock_execute, mock_exists):
path = '/some/path'
example_output = """image: 00000001
file format: raw
virtual size: 4.4M (4592640 bytes)
disk size: 4.4M
"""
mock_execute.return_value = (example_output, '')
self.assertEqual(4592640, disk.get_disk_size('/some/path'))
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
prlimit=images.QEMU_IMG_LIMITS)
mock_exists.assert_called_once_with(path)
def test_copy_image(self):
dst_fd, dst_path = tempfile.mkstemp()
try:
os.close(dst_fd)
src_fd, src_path = tempfile.mkstemp()
try:
with os.fdopen(src_fd, 'w') as fp:
fp.write('canary')
libvirt_utils.copy_image(src_path, dst_path)
with open(dst_path, 'r') as fp:
self.assertEqual(fp.read(), 'canary')
finally:
os.unlink(src_path)
finally:
os.unlink(dst_path)
def test_write_to_file(self):
dst_fd, dst_path = tempfile.mkstemp()
try:
os.close(dst_fd)
libvirt_utils.write_to_file(dst_path, 'hello')
with open(dst_path, 'r') as fp:
self.assertEqual(fp.read(), 'hello')
finally:
os.unlink(dst_path)
def test_write_to_file_with_umask(self):
dst_fd, dst_path = tempfile.mkstemp()
try:
os.close(dst_fd)
os.unlink(dst_path)
libvirt_utils.write_to_file(dst_path, 'hello', umask=0o277)
with open(dst_path, 'r') as fp:
self.assertEqual(fp.read(), 'hello')
mode = os.stat(dst_path).st_mode
self.assertEqual(mode & 0o277, 0)
finally:
os.unlink(dst_path)
@mock.patch.object(utils, 'execute')
def test_chown(self, mock_execute):
libvirt_utils.chown('/some/path', 'soren')
mock_execute.assert_called_once_with('chown', 'soren', '/some/path',
run_as_root=True)
@mock.patch.object(utils, 'execute')
def test_chown_for_id_maps(self, mock_execute):
id_maps = [vconfig.LibvirtConfigGuestUIDMap(),
vconfig.LibvirtConfigGuestUIDMap(),
vconfig.LibvirtConfigGuestGIDMap(),
vconfig.LibvirtConfigGuestGIDMap()]
id_maps[0].target = 10000
id_maps[0].count = 2000
id_maps[1].start = 2000
id_maps[1].target = 40000
id_maps[1].count = 2000
id_maps[2].target = 10000
id_maps[2].count = 2000
id_maps[3].start = 2000
id_maps[3].target = 40000
id_maps[3].count = 2000
libvirt_utils.chown_for_id_maps('/some/path', id_maps)
execute_args = ('nova-idmapshift', '-i',
'-u', '0:10000:2000,2000:40000:2000',
'-g', '0:10000:2000,2000:40000:2000',
'/some/path')
mock_execute.assert_called_once_with(*execute_args, run_as_root=True)
def _do_test_extract_snapshot(self, mock_execute, src_format='qcow2',
dest_format='raw', out_format='raw'):
libvirt_utils.extract_snapshot('/path/to/disk/image', src_format,
'/extracted/snap', dest_format)
mock_execute.assert_called_once_with(
'qemu-img', 'convert', '-f', src_format, '-O', out_format,
'/path/to/disk/image', '/extracted/snap')
@mock.patch.object(utils, 'execute')
def test_extract_snapshot_raw(self, mock_execute):
self._do_test_extract_snapshot(mock_execute)
@mock.patch.object(utils, 'execute')
def test_extract_snapshot_iso(self, mock_execute):
self._do_test_extract_snapshot(mock_execute, dest_format='iso')
@mock.patch.object(utils, 'execute')
def test_extract_snapshot_qcow2(self, mock_execute):
self._do_test_extract_snapshot(mock_execute,
dest_format='qcow2', out_format='qcow2')
@mock.patch.object(utils, 'execute')
def test_extract_snapshot_parallels(self, mock_execute):
self._do_test_extract_snapshot(mock_execute,
src_format='raw',
dest_format='ploop',
out_format='parallels')
def test_load_file(self):
dst_fd, dst_path = tempfile.mkstemp()
try:
os.close(dst_fd)
# We have a test for write_to_file. If that is sound, this suffices
libvirt_utils.write_to_file(dst_path, 'hello')
self.assertEqual(libvirt_utils.load_file(dst_path), 'hello')
finally:
os.unlink(dst_path)
def test_file_open(self):
dst_fd, dst_path = tempfile.mkstemp()
try:
os.close(dst_fd)
# We have a test for write_to_file. If that is sound, this suffices
libvirt_utils.write_to_file(dst_path, 'hello')
with libvirt_utils.file_open(dst_path, 'r') as fp:
self.assertEqual(fp.read(), 'hello')
finally:
os.unlink(dst_path)
def test_get_fs_info(self):
class FakeStatResult(object):
def __init__(self):
self.f_bsize = 4096
self.f_frsize = 4096
self.f_blocks = 2000
self.f_bfree = 1000
self.f_bavail = 900
self.f_files = 2000
self.f_ffree = 1000
self.f_favail = 900
self.f_flag = 4096
self.f_namemax = 255
self.path = None
def fake_statvfs(path):
self.path = path
return FakeStatResult()
self.stub_out('os.statvfs', fake_statvfs)
fs_info = libvirt_utils.get_fs_info('/some/file/path')
self.assertEqual('/some/file/path', self.path)
self.assertEqual(8192000, fs_info['total'])
self.assertEqual(3686400, fs_info['free'])
self.assertEqual(4096000, fs_info['used'])
@mock.patch('nova.virt.images.fetch_to_raw')
def test_fetch_image(self, mock_images):
context = 'opaque context'
target = '/tmp/targetfile'
image_id = '4'
libvirt_utils.fetch_image(context, target, image_id)
mock_images.assert_called_once_with(
context, image_id, target, max_size=0)
@mock.patch('nova.virt.images.fetch')
def test_fetch_initrd_image(self, mock_images):
_context = context.RequestContext(project_id=123,
project_name="aubergine",
user_id=456,
user_name="pie")
target = '/tmp/targetfile'
image_id = '4'
libvirt_utils.fetch_raw_image(_context, target, image_id)
mock_images.assert_called_once_with(
_context, image_id, target, max_size=0)
def test_fetch_raw_image(self):
def fake_execute(*cmd, **kwargs):
self.executes.append(cmd)
return None, None
def fake_rename(old, new):
self.executes.append(('mv', old, new))
def fake_unlink(path):
self.executes.append(('rm', path))
def fake_rm_on_error(path, remove=None):
self.executes.append(('rm', '-f', path))
def fake_qemu_img_info(path):
class FakeImgInfo(object):
pass
file_format = path.split('.')[-1]
if file_format == 'part':
file_format = path.split('.')[-2]
elif file_format == 'converted':
file_format = 'raw'
if 'backing' in path:
backing_file = 'backing'
else:
backing_file = None
if 'big' in path:
virtual_size = 2
else:
virtual_size = 1
FakeImgInfo.file_format = file_format
FakeImgInfo.backing_file = backing_file
FakeImgInfo.virtual_size = virtual_size
return FakeImgInfo()
self.stubs.Set(utils, 'execute', fake_execute)
self.stub_out('os.rename', fake_rename)
self.stub_out('os.unlink', fake_unlink)
self.stubs.Set(images, 'fetch', lambda *_, **__: None)
self.stubs.Set(images, 'qemu_img_info', fake_qemu_img_info)
self.stubs.Set(fileutils, 'delete_if_exists', fake_rm_on_error)
# Since the remove param of fileutils.remove_path_on_error()
# is initialized at load time, we must provide a wrapper
# that explicitly resets it to our fake delete_if_exists()
old_rm_path_on_error = fileutils.remove_path_on_error
f = functools.partial(old_rm_path_on_error, remove=fake_rm_on_error)
self.stubs.Set(fileutils, 'remove_path_on_error', f)
context = 'opaque context'
image_id = '4'
target = 't.qcow2'
self.executes = []
expected_commands = [('qemu-img', 'convert', '-O', 'raw',
't.qcow2.part', 't.qcow2.converted',
'-f', 'qcow2'),
('rm', 't.qcow2.part'),
('mv', 't.qcow2.converted', 't.qcow2')]
images.fetch_to_raw(context, image_id, target, max_size=1)
self.assertEqual(self.executes, expected_commands)
target = 't.raw'
self.executes = []
expected_commands = [('mv', 't.raw.part', 't.raw')]
images.fetch_to_raw(context, image_id, target)
self.assertEqual(self.executes, expected_commands)
target = 'backing.qcow2'
self.executes = []
expected_commands = [('rm', '-f', 'backing.qcow2.part')]
self.assertRaises(exception.ImageUnacceptable,
images.fetch_to_raw, context, image_id, target)
self.assertEqual(self.executes, expected_commands)
target = 'big.qcow2'
self.executes = []
expected_commands = [('rm', '-f', 'big.qcow2.part')]
self.assertRaises(exception.FlavorDiskSmallerThanImage,
images.fetch_to_raw,
context, image_id, target, max_size=1)
self.assertEqual(self.executes, expected_commands)
del self.executes
def test_get_disk_backing_file(self):
with_actual_path = False
def fake_execute(*args, **kwargs):
if with_actual_path:
return ("some: output\n"
"backing file: /foo/bar/baz (actual path: /a/b/c)\n"
"...: ...\n"), ''
else:
return ("some: output\n"
"backing file: /foo/bar/baz\n"
"...: ...\n"), ''
def return_true(*args, **kwargs):
return True
self.stubs.Set(utils, 'execute', fake_execute)
self.stub_out('os.path.exists', return_true)
out = libvirt_utils.get_disk_backing_file('')
self.assertEqual(out, 'baz')
with_actual_path = True
out = libvirt_utils.get_disk_backing_file('')
self.assertEqual(out, 'c')
def test_get_instance_path_at_destination(self):
instance = fake_instance.fake_instance_obj(None, name='fake_inst',
uuid=uuids.instance)
migrate_data = None
inst_path_at_dest = libvirt_utils.get_instance_path_at_destination(
instance, migrate_data)
expected_path = os.path.join(CONF.instances_path, instance['uuid'])
self.assertEqual(expected_path, inst_path_at_dest)
migrate_data = {}
inst_path_at_dest = libvirt_utils.get_instance_path_at_destination(
instance, migrate_data)
expected_path = os.path.join(CONF.instances_path, instance['uuid'])
self.assertEqual(expected_path, inst_path_at_dest)
migrate_data = objects.LibvirtLiveMigrateData(
instance_relative_path='fake_relative_path')
inst_path_at_dest = libvirt_utils.get_instance_path_at_destination(
instance, migrate_data)
expected_path = os.path.join(CONF.instances_path, 'fake_relative_path')
self.assertEqual(expected_path, inst_path_at_dest)
def test_get_arch(self):
image_meta = objects.ImageMeta.from_dict(
{'properties': {'architecture': "X86_64"}})
image_arch = libvirt_utils.get_arch(image_meta)
self.assertEqual(arch.X86_64, image_arch)
View Git Blame Copy Permalink