# Copyright (c) 2013 eNovance , 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. """Unit tests for image utils.""" import contextlib import tempfile import mock import mox from oslo.config import cfg from cinder import context from cinder import exception from cinder.image import image_utils from cinder.openstack.common import fileutils from cinder.openstack.common import processutils from cinder.openstack.common import units from cinder import test from cinder import utils from cinder.volume import utils as volume_utils CONF = cfg.CONF class FakeImageService: def __init__(self): self._imagedata = {} def download(self, context, image_id, data): self.show(context, image_id) data.write(self._imagedata.get(image_id, '')) def show(self, context, image_id): return {'size': 2 * units.Gi, 'disk_format': 'qcow2', 'container_format': 'bare'} def update(self, context, image_id, metadata, path): pass class TestUtils(test.TestCase): TEST_IMAGE_ID = 321 TEST_DEV_PATH = "/dev/ether/fake_dev" def setUp(self): super(TestUtils, self).setUp() self._mox = mox.Mox() self._image_service = FakeImageService() self.addCleanup(self._mox.UnsetStubs) def test_resize_image(self): mox = self._mox mox.StubOutWithMock(utils, 'execute') TEST_IMG_SOURCE = 'boobar.img' TEST_IMG_SIZE_IN_GB = 1 utils.execute('qemu-img', 'resize', TEST_IMG_SOURCE, '%sG' % TEST_IMG_SIZE_IN_GB, run_as_root=False) mox.ReplayAll() image_utils.resize_image(TEST_IMG_SOURCE, TEST_IMG_SIZE_IN_GB) mox.VerifyAll() @mock.patch('os.stat') def test_convert_image(self, mock_stat): mox = self._mox mox.StubOutWithMock(utils, 'execute') mox.StubOutWithMock(utils, 'is_blk_device') TEST_OUT_FORMAT = 'vmdk' TEST_SOURCE = 'img/qemu.img' TEST_DEST = '/img/vmware.vmdk' utils.is_blk_device(TEST_DEST).AndReturn(True) utils.execute('dd', 'count=0', 'if=img/qemu.img', 'of=/img/vmware.vmdk', 'oflag=direct', run_as_root=True) utils.execute( 'qemu-img', 'convert', '-t', 'none', '-O', TEST_OUT_FORMAT, TEST_SOURCE, TEST_DEST, run_as_root=True) mox.ReplayAll() image_utils.convert_image(TEST_SOURCE, TEST_DEST, TEST_OUT_FORMAT) mox.VerifyAll() def test_qemu_img_info(self): TEST_PATH = "img/qemu.qcow2" TEST_RETURN = "image: qemu.qcow2\n"\ "backing_file: qemu.qcow2 (actual path: qemu.qcow2)\n"\ "file_format: qcow2\n"\ "virtual_size: 50M (52428800 bytes)\n"\ "cluster_size: 65536\n"\ "disk_size: 196K (200704 bytes)\n"\ "Snapshot list:\n"\ "ID TAG VM SIZE DATE VM CLOCK\n"\ "1 snap1 1.7G 2011-10-04 19:04:00 32:06:34.974" TEST_STR = "image: qemu.qcow2\n"\ "file_format: qcow2\n"\ "virtual_size: 52428800\n"\ "disk_size: 200704\n"\ "cluster_size: 65536\n"\ "backing_file: qemu.qcow2\n"\ "snapshots: [{'date': '2011-10-04', "\ "'vm_clock': '19:04:00 32:06:34.974', "\ "'vm_size': '1.7G', 'tag': 'snap1', 'id': '1'}]" mox = self._mox mox.StubOutWithMock(utils, 'execute') utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', TEST_PATH, run_as_root=True).AndReturn( (TEST_RETURN, 'ignored')) mox.ReplayAll() inf = image_utils.qemu_img_info(TEST_PATH) self.assertEqual(inf.image, 'qemu.qcow2') self.assertEqual(inf.backing_file, 'qemu.qcow2') self.assertEqual(inf.file_format, 'qcow2') self.assertEqual(inf.virtual_size, 52428800) self.assertEqual(inf.cluster_size, 65536) self.assertEqual(inf.disk_size, 200704) self.assertEqual(inf.snapshots[0]['id'], '1') self.assertEqual(inf.snapshots[0]['tag'], 'snap1') self.assertEqual(inf.snapshots[0]['vm_size'], '1.7G') self.assertEqual(inf.snapshots[0]['date'], '2011-10-04') self.assertEqual(inf.snapshots[0]['vm_clock'], '19:04:00 32:06:34.974') self.assertEqual(str(inf), TEST_STR) def test_qemu_img_info_alt(self): """Test a slightly different variation of qemu-img output. (Based on Fedora 19's qemu-img 1.4.2.) """ TEST_PATH = "img/qemu.qcow2" TEST_RETURN = "image: qemu.qcow2\n"\ "backing file: qemu.qcow2 (actual path: qemu.qcow2)\n"\ "file format: qcow2\n"\ "virtual size: 50M (52428800 bytes)\n"\ "cluster_size: 65536\n"\ "disk size: 196K (200704 bytes)\n"\ "Snapshot list:\n"\ "ID TAG VM SIZE DATE VM CLOCK\n"\ "1 snap1 1.7G 2011-10-04 19:04:00 32:06:34.974" TEST_STR = "image: qemu.qcow2\n"\ "file_format: qcow2\n"\ "virtual_size: 52428800\n"\ "disk_size: 200704\n"\ "cluster_size: 65536\n"\ "backing_file: qemu.qcow2\n"\ "snapshots: [{'date': '2011-10-04', "\ "'vm_clock': '19:04:00 32:06:34.974', "\ "'vm_size': '1.7G', 'tag': 'snap1', 'id': '1'}]" mox = self._mox mox.StubOutWithMock(utils, 'execute') cmd = ['env', 'LC_ALL=C', 'qemu-img', 'info', TEST_PATH] utils.execute(*cmd, run_as_root=True).AndReturn( (TEST_RETURN, 'ignored')) mox.ReplayAll() inf = image_utils.qemu_img_info(TEST_PATH) self.assertEqual(inf.image, 'qemu.qcow2') self.assertEqual(inf.backing_file, 'qemu.qcow2') self.assertEqual(inf.file_format, 'qcow2') self.assertEqual(inf.virtual_size, 52428800) self.assertEqual(inf.cluster_size, 65536) self.assertEqual(inf.disk_size, 200704) self.assertEqual(inf.snapshots[0]['id'], '1') self.assertEqual(inf.snapshots[0]['tag'], 'snap1') self.assertEqual(inf.snapshots[0]['vm_size'], '1.7G') self.assertEqual(inf.snapshots[0]['date'], '2011-10-04') self.assertEqual(inf.snapshots[0]['vm_clock'], '19:04:00 32:06:34.974') self.assertEqual(str(inf), TEST_STR) def _test_fetch_to_raw(self, has_qemu=True, src_inf=None, dest_inf=None, bps_limit=0): mox = self._mox mox.StubOutWithMock(image_utils, 'create_temporary_file') mox.StubOutWithMock(utils, 'execute') mox.StubOutWithMock(image_utils, 'fetch') mox.StubOutWithMock(volume_utils, 'setup_blkio_cgroup') mox.StubOutWithMock(utils, 'is_blk_device') TEST_INFO = ("image: qemu.qcow2\n" "file format: raw\n" "virtual size: 0 (0 bytes)\n" "disk size: 0") utils.is_blk_device(self.TEST_DEV_PATH).AndReturn(True) CONF.set_override('volume_copy_bps_limit', bps_limit) image_utils.create_temporary_file().AndReturn(self.TEST_DEV_PATH) test_qemu_img = utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', self.TEST_DEV_PATH, run_as_root=True) if has_qemu: test_qemu_img.AndReturn((TEST_INFO, 'ignored')) image_utils.fetch(context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None) else: test_qemu_img.AndRaise(processutils.ProcessExecutionError()) if has_qemu and src_inf: utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', self.TEST_DEV_PATH, run_as_root=True).AndReturn( (src_inf, 'ignored')) if has_qemu and dest_inf: if bps_limit: prefix = ('cgexec', '-g', 'blkio:test') else: prefix = () utils.execute('dd', 'count=0', 'if=/dev/ether/fake_dev', 'of=/dev/ether/fake_dev', 'oflag=direct', run_as_root=True) cmd = prefix + ('qemu-img', 'convert', '-t', 'none', '-O', 'raw', self.TEST_DEV_PATH, self.TEST_DEV_PATH) volume_utils.setup_blkio_cgroup( self.TEST_DEV_PATH, self.TEST_DEV_PATH, bps_limit).AndReturn(prefix) utils.execute(*cmd, run_as_root=True) utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', self.TEST_DEV_PATH, run_as_root=True).AndReturn( (dest_inf, 'ignored')) self._mox.ReplayAll() @mock.patch('os.stat') def test_fetch_to_raw(self, mock_stat): SRC_INFO = ("image: qemu.qcow2\n" "file_format: qcow2 \n" "virtual_size: 50M (52428800 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)") DST_INFO = ("image: qemu.raw\n" "file_format: raw\n" "virtual_size: 50M (52428800 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)\n") self._test_fetch_to_raw(src_inf=SRC_INFO, dest_inf=DST_INFO) image_utils.fetch_to_raw(context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, mox.IgnoreArg()) self._mox.VerifyAll() @mock.patch('os.stat') def test_fetch_to_raw_with_bps_limit(self, mock_stat): SRC_INFO = ("image: qemu.qcow2\n" "file_format: qcow2 \n" "virtual_size: 50M (52428800 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)") DST_INFO = ("image: qemu.raw\n" "file_format: raw\n" "virtual_size: 50M (52428800 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)\n") self._test_fetch_to_raw(src_inf=SRC_INFO, dest_inf=DST_INFO, bps_limit=1048576) image_utils.fetch_to_raw(context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, mox.IgnoreArg()) self._mox.VerifyAll() def test_fetch_to_raw_no_qemu_img(self): self._test_fetch_to_raw(has_qemu=False) self.assertRaises(exception.ImageUnacceptable, image_utils.fetch_to_raw, context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, mox.IgnoreArg()) def test_fetch_to_raw_on_error_parsing_failed(self): SRC_INFO_NO_FORMAT = ("image: qemu.qcow2\n" "virtual_size: 50M (52428800 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)") self._test_fetch_to_raw(src_inf=SRC_INFO_NO_FORMAT) self.assertRaises(exception.ImageUnacceptable, image_utils.fetch_to_raw, context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, mox.IgnoreArg()) def test_fetch_to_raw_on_error_backing_file(self): SRC_INFO_BACKING_FILE = ("image: qemu.qcow2\n" "backing_file: qemu.qcow2\n" "file_format: qcow2 \n" "virtual_size: 50M (52428800 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)") self._test_fetch_to_raw(src_inf=SRC_INFO_BACKING_FILE) self.assertRaises(exception.ImageUnacceptable, image_utils.fetch_to_raw, context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, mox.IgnoreArg()) @mock.patch('os.stat') def test_fetch_to_raw_on_error_not_convert_to_raw(self, mock_stat): IMG_INFO = ("image: qemu.qcow2\n" "file_format: qcow2 \n" "virtual_size: 50M (52428800 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)") self._test_fetch_to_raw(src_inf=IMG_INFO, dest_inf=IMG_INFO) self.assertRaises(exception.ImageUnacceptable, image_utils.fetch_to_raw, context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, mox.IgnoreArg()) def test_fetch_to_raw_on_error_image_size(self): TEST_VOLUME_SIZE = 1 SRC_INFO = ("image: qemu.qcow2\n" "file_format: qcow2 \n" "virtual_size: 2G (2147483648 bytes)\n" "cluster_size: 65536\n" "disk_size: 196K (200704 bytes)") self._test_fetch_to_raw(src_inf=SRC_INFO) self.assertRaises(exception.ImageUnacceptable, image_utils.fetch_to_raw, context, self._image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, mox.IgnoreArg(), size=TEST_VOLUME_SIZE) def _test_fetch_verify_image(self, qemu_info, volume_size=1): fake_image_service = FakeImageService() mox = self._mox mox.StubOutWithMock(image_utils, 'fetch') mox.StubOutWithMock(utils, 'execute') image_utils.fetch(context, fake_image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None) utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', self.TEST_DEV_PATH, run_as_root=True).AndReturn( (qemu_info, 'ignored')) self._mox.ReplayAll() self.assertRaises(exception.ImageUnacceptable, image_utils.fetch_verify_image, context, fake_image_service, self.TEST_IMAGE_ID, self.TEST_DEV_PATH, size=volume_size) def test_fetch_verify_image_with_backing_file(self): TEST_RETURN = "image: qemu.qcow2\n"\ "backing_file: qemu.qcow2 (actual path: qemu.qcow2)\n"\ "file_format: qcow2\n"\ "virtual_size: 50M (52428800 bytes)\n"\ "cluster_size: 65536\n"\ "disk_size: 196K (200704 bytes)\n"\ "Snapshot list:\n"\ "ID TAG VM SIZE DATE VM CLOCK\n"\ "1 snap1 1.7G 2011-10-04 19:04:00 32:06:34.974" self._test_fetch_verify_image(TEST_RETURN) def test_fetch_verify_image_without_file_format(self): TEST_RETURN = "image: qemu.qcow2\n"\ "virtual_size: 50M (52428800 bytes)\n"\ "cluster_size: 65536\n"\ "disk_size: 196K (200704 bytes)\n"\ "Snapshot list:\n"\ "ID TAG VM SIZE DATE VM CLOCK\n"\ "1 snap1 1.7G 2011-10-04 19:04:00 32:06:34.974" self._test_fetch_verify_image(TEST_RETURN) def test_fetch_verify_image_image_size(self): TEST_RETURN = "image: qemu.qcow2\n"\ "file_format: qcow2\n"\ "virtual_size: 2G (2147483648 bytes)\n"\ "cluster_size: 65536\n"\ "disk_size: 196K (200704 bytes)\n"\ "Snapshot list:\n"\ "ID TAG VM SIZE DATE VM CLOCK\n"\ "1 snap1 1.7G 2011-10-04 19:04:00 32:06:34.974" self._test_fetch_verify_image(TEST_RETURN) @mock.patch('os.stat') def test_upload_volume(self, mock_stat, bps_limit=0): image_meta = {'id': 1, 'disk_format': 'qcow2'} TEST_RET = "image: qemu.qcow2\n"\ "file_format: qcow2 \n"\ "virtual_size: 50M (52428800 bytes)\n"\ "cluster_size: 65536\n"\ "disk_size: 196K (200704 bytes)" if bps_limit: CONF.set_override('volume_copy_bps_limit', bps_limit) prefix = ('cgexec', '-g', 'blkio:test') else: prefix = () cmd = prefix + ('qemu-img', 'convert', '-O', 'qcow2', mox.IgnoreArg(), mox.IgnoreArg()) m = self._mox m.StubOutWithMock(utils, 'execute') m.StubOutWithMock(volume_utils, 'setup_blkio_cgroup') volume_utils.setup_blkio_cgroup(mox.IgnoreArg(), mox.IgnoreArg(), bps_limit).AndReturn(prefix) utils.execute(*cmd, run_as_root=True) utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', mox.IgnoreArg(), run_as_root=True).AndReturn( (TEST_RET, 'ignored')) m.ReplayAll() image_utils.upload_volume(context, FakeImageService(), image_meta, '/dev/loop1') m.VerifyAll() @mock.patch('os.stat') def test_upload_volume_with_bps_limit(self, mock_stat): bps_limit = 1048576 image_meta = {'id': 1, 'disk_format': 'qcow2'} TEST_RET = "image: qemu.qcow2\n"\ "file_format: qcow2 \n"\ "virtual_size: 50M (52428800 bytes)\n"\ "cluster_size: 65536\n"\ "disk_size: 196K (200704 bytes)" CONF.set_override('volume_copy_bps_limit', bps_limit) prefix = ('cgexec', '-g', 'blkio:test') cmd = prefix + ('qemu-img', 'convert', '-O', 'qcow2', mox.IgnoreArg(), mox.IgnoreArg()) m = self._mox m.StubOutWithMock(utils, 'execute') m.StubOutWithMock(volume_utils, 'setup_blkio_cgroup') m.StubOutWithMock(volume_utils, 'check_for_odirect_support') volume_utils.setup_blkio_cgroup(mox.IgnoreArg(), mox.IgnoreArg(), bps_limit).AndReturn(prefix) utils.execute(*cmd, run_as_root=True) utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', mox.IgnoreArg(), run_as_root=True).AndReturn( (TEST_RET, 'ignored')) m.ReplayAll() image_utils.upload_volume(context, FakeImageService(), image_meta, '/dev/loop1') m.VerifyAll() def test_upload_volume_with_raw_image(self): image_meta = {'id': 1, 'disk_format': 'raw'} mox = self._mox mox.StubOutWithMock(image_utils, 'convert_image') mox.ReplayAll() with tempfile.NamedTemporaryFile() as f: image_utils.upload_volume(context, FakeImageService(), image_meta, f.name) mox.VerifyAll() @mock.patch('os.stat') def test_upload_volume_on_error(self, mock_stat): image_meta = {'id': 1, 'disk_format': 'qcow2'} TEST_RET = "image: qemu.vhd\n"\ "file_format: vhd \n"\ "virtual_size: 50M (52428800 bytes)\n"\ "cluster_size: 65536\n"\ "disk_size: 196K (200704 bytes)" m = self._mox m.StubOutWithMock(utils, 'execute') m.StubOutWithMock(volume_utils, 'check_for_odirect_support') utils.execute('qemu-img', 'convert', '-O', 'qcow2', mox.IgnoreArg(), mox.IgnoreArg(), run_as_root=True) utils.execute( 'env', 'LC_ALL=C', 'qemu-img', 'info', mox.IgnoreArg(), run_as_root=True).AndReturn( (TEST_RET, 'ignored')) m.ReplayAll() self.assertRaises(exception.ImageUnacceptable, image_utils.upload_volume, context, FakeImageService(), image_meta, '/dev/loop1') m.VerifyAll() class TestExtractTo(test.TestCase): def test_extract_to_calls_tar(self): mox = self.mox mox.StubOutWithMock(utils, 'execute') utils.execute( 'tar', '-xzf', 'archive.tgz', '-C', 'targetpath').AndReturn( ('ignored', 'ignored')) mox.ReplayAll() image_utils.extract_targz('archive.tgz', 'targetpath') mox.VerifyAll() class TestSetVhdParent(test.TestCase): def test_vhd_util_call(self): mox = self.mox mox.StubOutWithMock(utils, 'execute') utils.execute( 'vhd-util', 'modify', '-n', 'child', '-p', 'parent').AndReturn( ('ignored', 'ignored')) mox.ReplayAll() image_utils.set_vhd_parent('child', 'parent') mox.VerifyAll() class TestFixVhdChain(test.TestCase): def test_empty_chain(self): mox = self.mox mox.StubOutWithMock(image_utils, 'set_vhd_parent') mox.ReplayAll() image_utils.fix_vhd_chain([]) def test_single_vhd_file_chain(self): mox = self.mox mox.StubOutWithMock(image_utils, 'set_vhd_parent') mox.ReplayAll() image_utils.fix_vhd_chain(['0.vhd']) def test_chain_with_two_elements(self): mox = self.mox mox.StubOutWithMock(image_utils, 'set_vhd_parent') image_utils.set_vhd_parent('0.vhd', '1.vhd') mox.ReplayAll() image_utils.fix_vhd_chain(['0.vhd', '1.vhd']) class TestGetSize(test.TestCase): def test_vhd_util_call(self): mox = self.mox mox.StubOutWithMock(utils, 'execute') utils.execute( 'vhd-util', 'query', '-n', 'vhdfile', '-v').AndReturn( ('1024', 'ignored')) mox.ReplayAll() result = image_utils.get_vhd_size('vhdfile') mox.VerifyAll() self.assertEqual(1024, result) class TestResize(test.TestCase): def test_vhd_util_call(self): mox = self.mox mox.StubOutWithMock(utils, 'execute') utils.execute( 'vhd-util', 'resize', '-n', 'vhdfile', '-s', '1024', '-j', 'journal').AndReturn(('ignored', 'ignored')) mox.ReplayAll() image_utils.resize_vhd('vhdfile', 1024, 'journal') mox.VerifyAll() class TestCoalesce(test.TestCase): def test_vhd_util_call(self): mox = self.mox mox.StubOutWithMock(utils, 'execute') utils.execute( 'vhd-util', 'coalesce', '-n', 'vhdfile' ).AndReturn(('ignored', 'ignored')) mox.ReplayAll() image_utils.coalesce_vhd('vhdfile') mox.VerifyAll() @contextlib.contextmanager def fake_context(return_value): yield return_value class TestTemporaryFile(test.TestCase): def test_file_unlinked(self): mox = self.mox mox.StubOutWithMock(image_utils, 'create_temporary_file') mox.StubOutWithMock(fileutils, 'delete_if_exists') image_utils.create_temporary_file().AndReturn('somefile') fileutils.delete_if_exists('somefile') mox.ReplayAll() with image_utils.temporary_file(): pass def test_file_unlinked_on_error(self): mox = self.mox mox.StubOutWithMock(image_utils, 'create_temporary_file') mox.StubOutWithMock(fileutils, 'delete_if_exists') image_utils.create_temporary_file().AndReturn('somefile') fileutils.delete_if_exists('somefile') mox.ReplayAll() def sut(): with image_utils.temporary_file(): raise test.TestingException() self.assertRaises(test.TestingException, sut) class TestCoalesceChain(test.TestCase): def test_single_vhd(self): mox = self.mox mox.StubOutWithMock(image_utils, 'get_vhd_size') mox.StubOutWithMock(image_utils, 'resize_vhd') mox.StubOutWithMock(image_utils, 'coalesce_vhd') mox.ReplayAll() result = image_utils.coalesce_chain(['0.vhd']) mox.VerifyAll() self.assertEqual('0.vhd', result) def test_chain_of_two_vhds(self): self.mox.StubOutWithMock(image_utils, 'get_vhd_size') self.mox.StubOutWithMock(image_utils, 'temporary_dir') self.mox.StubOutWithMock(image_utils, 'resize_vhd') self.mox.StubOutWithMock(image_utils, 'coalesce_vhd') self.mox.StubOutWithMock(image_utils, 'temporary_file') image_utils.get_vhd_size('0.vhd').AndReturn(1024) image_utils.temporary_dir().AndReturn(fake_context('tdir')) image_utils.resize_vhd('1.vhd', 1024, 'tdir/vhd-util-resize-journal') image_utils.coalesce_vhd('0.vhd') self.mox.ReplayAll() result = image_utils.coalesce_chain(['0.vhd', '1.vhd']) self.mox.VerifyAll() self.assertEqual('1.vhd', result) class TestDiscoverChain(test.TestCase): def test_discovery_calls(self): mox = self.mox mox.StubOutWithMock(image_utils, 'file_exist') image_utils.file_exist('some/path/0.vhd').AndReturn(True) image_utils.file_exist('some/path/1.vhd').AndReturn(True) image_utils.file_exist('some/path/2.vhd').AndReturn(False) mox.ReplayAll() result = image_utils.discover_vhd_chain('some/path') mox.VerifyAll() self.assertEqual( ['some/path/0.vhd', 'some/path/1.vhd'], result) class TestXenServerImageToCoalescedVhd(test.TestCase): def test_calls(self): mox = self.mox mox.StubOutWithMock(image_utils, 'temporary_dir') mox.StubOutWithMock(image_utils, 'extract_targz') mox.StubOutWithMock(image_utils, 'discover_vhd_chain') mox.StubOutWithMock(image_utils, 'fix_vhd_chain') mox.StubOutWithMock(image_utils, 'coalesce_chain') mox.StubOutWithMock(image_utils.os, 'unlink') mox.StubOutWithMock(fileutils, 'delete_if_exists') mox.StubOutWithMock(image_utils, 'rename_file') image_utils.temporary_dir().AndReturn(fake_context('somedir')) image_utils.extract_targz('image', 'somedir') image_utils.discover_vhd_chain('somedir').AndReturn( ['somedir/0.vhd', 'somedir/1.vhd']) image_utils.fix_vhd_chain(['somedir/0.vhd', 'somedir/1.vhd']) image_utils.coalesce_chain( ['somedir/0.vhd', 'somedir/1.vhd']).AndReturn('somedir/1.vhd') fileutils.delete_if_exists('image') image_utils.rename_file('somedir/1.vhd', 'image') mox.ReplayAll() image_utils.replace_xenserver_image_with_coalesced_vhd('image') mox.VerifyAll()