755 lines
29 KiB
Python
755 lines
29 KiB
Python
# 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 exception
|
|
from nova import test
|
|
from nova.tests.unit import fake_instance
|
|
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('os.path.exists', return_value=True)
|
|
@mock.patch('nova.utils.execute')
|
|
def test_get_disk_type(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, '')
|
|
disk_type = libvirt_utils.get_disk_type(path)
|
|
mock_execute.assert_called_once_with('env', 'LC_ALL=C', 'LANG=C',
|
|
'qemu-img', 'info', path)
|
|
mock_exists.assert_called_once_with(path)
|
|
self.assertEqual('raw', disk_type)
|
|
|
|
@mock.patch('nova.utils.execute')
|
|
def test_copy_image_local_cp(self, mock_execute):
|
|
libvirt_utils.copy_image('src', 'dest')
|
|
mock_execute.assert_called_once_with('cp', 'src', 'dest')
|
|
|
|
_rsync_call = functools.partial(mock.call,
|
|
'rsync', '--sparse', '--compress',
|
|
on_execute=None, on_completion=None)
|
|
|
|
@mock.patch('nova.utils.execute')
|
|
def test_copy_image_rsync(self, mock_execute):
|
|
libvirt_utils.copy_image('src', 'dest', host='host')
|
|
|
|
mock_execute.assert_has_calls([
|
|
self._rsync_call('--dry-run', 'src', 'host:dest'),
|
|
self._rsync_call('src', 'host:dest'),
|
|
])
|
|
self.assertEqual(2, mock_execute.call_count)
|
|
|
|
@mock.patch('nova.utils.execute')
|
|
def test_copy_image_scp(self, mock_execute):
|
|
mock_execute.side_effect = [
|
|
processutils.ProcessExecutionError,
|
|
mock.DEFAULT,
|
|
]
|
|
|
|
libvirt_utils.copy_image('src', 'dest', host='host')
|
|
|
|
mock_execute.assert_has_calls([
|
|
self._rsync_call('--dry-run', 'src', 'host:dest'),
|
|
mock.call('scp', 'src', 'host:dest',
|
|
on_execute=None, on_completion=None),
|
|
])
|
|
self.assertEqual(2, mock_execute.call_count)
|
|
|
|
@mock.patch('nova.utils.execute')
|
|
def test_copy_image_rsync_ipv6(self, mock_execute):
|
|
libvirt_utils.copy_image('src', 'dest', host='2600::')
|
|
|
|
mock_execute.assert_has_calls([
|
|
self._rsync_call('--dry-run', 'src', '[2600::]:dest'),
|
|
self._rsync_call('src', '[2600::]:dest'),
|
|
])
|
|
self.assertEqual(2, mock_execute.call_count)
|
|
|
|
@mock.patch('nova.utils.execute')
|
|
def test_copy_image_scp_ipv6(self, mock_execute):
|
|
mock_execute.side_effect = [
|
|
processutils.ProcessExecutionError,
|
|
mock.DEFAULT,
|
|
]
|
|
|
|
libvirt_utils.copy_image('src', 'dest', host='2600::')
|
|
|
|
mock_execute.assert_has_calls([
|
|
self._rsync_call('--dry-run', 'src', '[2600::]:dest'),
|
|
mock.call('scp', 'src', '[2600::]:dest',
|
|
on_execute=None, on_completion=None),
|
|
])
|
|
self.assertEqual(2, mock_execute.call_count)
|
|
|
|
@mock.patch('os.path.exists', return_value=True)
|
|
def test_disk_type(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(p)
|
|
self.assertEqual('lvm', d_type)
|
|
|
|
# Try rbd detection
|
|
d_type = libvirt_utils.get_disk_type('rbd:pool/instance')
|
|
self.assertEqual('rbd', d_type)
|
|
|
|
# Try the other types
|
|
template_output = """image: %(path)s
|
|
file format: %(format)s
|
|
virtual size: 64M (67108864 bytes)
|
|
cluster_size: 65536
|
|
disk size: 96K
|
|
"""
|
|
path = '/myhome/disk.config'
|
|
for f in ['raw', 'qcow2']:
|
|
output = template_output % ({
|
|
'format': f,
|
|
'path': path,
|
|
})
|
|
with mock.patch('nova.utils.execute',
|
|
return_value=(output, '')) as mock_execute:
|
|
d_type = libvirt_utils.get_disk_type(path)
|
|
mock_execute.assert_called_once_with(
|
|
'env', 'LC_ALL=C', 'LANG=C',
|
|
'qemu-img', 'info', path)
|
|
self.assertEqual(f, 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)
|
|
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)
|
|
|
|
@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)
|
|
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)
|
|
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('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)
|
|
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)
|
|
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)
|
|
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'),),
|
|
(('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)
|
|
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,
|
|
dest_format='raw', out_format='raw'):
|
|
libvirt_utils.extract_snapshot('/path/to/disk/image', 'qcow2',
|
|
'/extracted/snap', dest_format)
|
|
mock_execute.assert_called_once_with(
|
|
'qemu-img', 'convert', '-f', 'qcow2', '-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')
|
|
|
|
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.stubs.Set(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'
|
|
user_id = 'fake'
|
|
project_id = 'fake'
|
|
libvirt_utils.fetch_image(context, target, image_id,
|
|
user_id, project_id)
|
|
mock_images.assert_called_once_with(
|
|
context, image_id, target, user_id, project_id,
|
|
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.stubs.Set(os, 'rename', fake_rename)
|
|
self.stubs.Set(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'
|
|
user_id = 'fake'
|
|
project_id = 'fake'
|
|
|
|
target = 't.qcow2'
|
|
self.executes = []
|
|
expected_commands = [('qemu-img', 'convert', '-O', 'raw',
|
|
't.qcow2.part', 't.qcow2.converted'),
|
|
('rm', 't.qcow2.part'),
|
|
('mv', 't.qcow2.converted', 't.qcow2')]
|
|
images.fetch_to_raw(context, image_id, target, user_id, project_id,
|
|
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, user_id, project_id)
|
|
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, user_id, project_id)
|
|
self.assertEqual(self.executes, expected_commands)
|
|
|
|
target = 'big.qcow2'
|
|
self.executes = []
|
|
expected_commands = [('rm', '-f', 'big.qcow2.part')]
|
|
self.assertRaises(exception.FlavorDiskTooSmall,
|
|
images.fetch_to_raw,
|
|
context, image_id, target, user_id, project_id,
|
|
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.stubs.Set(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='fake_uuid')
|
|
|
|
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 = dict(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 = {'properties': {'architecture': "X86_64"}}
|
|
image_arch = libvirt_utils.get_arch(image_meta)
|
|
self.assertEqual(arch.X86_64, image_arch)
|