#!/usr/bin/env python3
#
# Tests for internal snapshot.
#
# Copyright (C) 2013 IBM, Inc.
#
# Based on 055.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import time
import os
import iotests
from iotests import qemu_img, qemu_io
test_drv_base_name = 'drive'
class ImageSnapshotTestCase(iotests.QMPTestCase):
image_len = 120 * 1024 * 1024 # MB
def __init__(self, *args):
self.expect = []
super(ImageSnapshotTestCase, self).__init__(*args)
def _setUp(self, test_img_base_name, image_num):
self.vm = iotests.VM()
for i in range(0, image_num):
filename = '%s%d' % (test_img_base_name, i)
img = os.path.join(iotests.test_dir, filename)
device = '%s%d' % (test_drv_base_name, i)
qemu_img('create', '-f', iotests.imgfmt, img, str(self.image_len))
self.vm.add_drive(img)
self.expect.append({'image': img, 'device': device,
'snapshots': [],
'snapshots_name_counter': 0})
self.vm.launch()
def tearDown(self):
self.vm.shutdown()
for dev_expect in self.expect:
os.remove(dev_expect['image'])
def createSnapshotInTransaction(self, snapshot_num, abort = False):
actions = []
for dev_expect in self.expect:
num = dev_expect['snapshots_name_counter']
for j in range(0, snapshot_num):
name = '%s_sn%d' % (dev_expect['device'], num)
num = num + 1
if abort == False:
dev_expect['snapshots'].append({'name': name})
dev_expect['snapshots_name_counter'] = num
actions.append({
'type': 'blockdev-snapshot-internal-sync',
'data': { 'device': dev_expect['device'],
'name': name },
})
if abort == True:
actions.append({
'type': 'abort',
'data': {},
})
result = self.vm.qmp('transaction', actions = actions)
if abort == True:
self.assert_qmp(result, 'error/class', 'GenericError')
else:
self.assert_qmp(result, 'return', {})
def verifySnapshotInfo(self):
result = self.vm.qmp('query-block')
# Verify each expected result
for dev_expect in self.expect:
# 1. Find the returned image value and snapshot info
image_result = None
for device in result['return']:
if device['device'] == dev_expect['device']:
image_result = device['inserted']['image']
break
self.assertTrue(image_result != None)
# Do not consider zero snapshot case now
sn_list_result = image_result['snapshots']
sn_list_expect = dev_expect['snapshots']
# 2. Verify it with expect
self.assertTrue(len(sn_list_result) == len(sn_list_expect))
for sn_expect in sn_list_expect:
sn_result = None
for sn in sn_list_result:
if sn_expect['name'] == sn['name']:
sn_result = sn
break
self.assertTrue(sn_result != None)
# Fill in the detail info
sn_expect.update(sn_result)
def deleteSnapshot(self, device, id = None, name = None):
sn_list_expect = None
sn_expect = None
self.assertTrue(id != None or name != None)
# Fill in the detail info include ID
self.verifySnapshotInfo()
#find the expected snapshot list
for dev_expect in self.expect:
if dev_expect['device'] == device:
sn_list_expect = dev_expect['snapshots']
break
self.assertTrue(sn_list_expect != None)
if id != None and name != None:
for sn in sn_list_expect:
if sn['id'] == id and sn['name'] == name:
sn_expect = sn
result = \
self.vm.qmp('blockdev-snapshot-delete-internal-sync',
device = device,
id = id,
name = name)
break
elif id != None:
for sn in sn_list_expect:
if sn['id'] == id:
sn_expect = sn
result = \
self.vm.qmp('blockdev-snapshot-delete-internal-sync',
device = device,
id = id)
break
else:
for sn in sn_list_expect:
if sn['name'] == name:
sn_expect = sn
result = \
self.vm.qmp('blockdev-snapshot-delete-internal-sync',
device = device,
name = name)
break
self.assertTrue(sn_expect != None)
self.assert_qmp(result, 'return', sn_expect)
sn_list_expect.remove(sn_expect)
class TestSingleTransaction(ImageSnapshotTestCase):
def setUp(self):
self._setUp('test_a.img', 1)
def test_create(self):
self.createSnapshotInTransaction(1)
self.verifySnapshotInfo()
def test_error_name_empty(self):
actions = [{'type': 'blockdev-snapshot-internal-sync',
'data': { 'device': self.expect[0]['device'],
'name': '' },
}]
result = self.vm.qmp('transaction', actions = actions)
self.assert_qmp(result, 'error/class', 'GenericError')
def test_error_device(self):
actions = [{'type': 'blockdev-snapshot-internal-sync',
'data': { 'device': 'drive_error',
'name': 'a' },
}]
result = self.vm.qmp('transaction', actions = actions)
self.assert_qmp(result, 'error/class', 'GenericError')
def test_error_exist(self):
self.createSnapshotInTransaction(1)
self.verifySnapshotInfo()
actions = [{'type': 'blockdev-snapshot-internal-sync',
'data': { 'device': self.expect[0]['device'],
'name': self.expect[0]['snapshots'][0] },
}]
result = self.vm.qmp('transaction', actions = actions)
self.assert_qmp(result, 'error/class', 'GenericError')
class TestMultipleTransaction(ImageSnapshotTestCase):
def setUp(self):
self._setUp('test_b.img', 2)
def test_create(self):
self.createSnapshotInTransaction(3)
self.verifySnapshotInfo()
def test_abort(self):
self.createSnapshotInTransaction(2)
self.verifySnapshotInfo()
self.createSnapshotInTransaction(3, abort = True)
self.verifySnapshotInfo()
class TestSnapshotDelete(ImageSnapshotTestCase):
def setUp(self):
self._setUp('test_c.img', 1)
def test_delete_with_id(self):
self.createSnapshotInTransaction(2)
self.verifySnapshotInfo()
self.deleteSnapshot(self.expect[0]['device'],
id = self.expect[0]['snapshots'][0]['id'])
self.verifySnapshotInfo()
def test_delete_with_name(self):
self.createSnapshotInTransaction(3)
self.verifySnapshotInfo()
self.deleteSnapshot(self.expect[0]['device'],
name = self.expect[0]['snapshots'][1]['name'])
self.verifySnapshotInfo()
def test_delete_with_id_and_name(self):
self.createSnapshotInTransaction(4)
self.verifySnapshotInfo()
self.deleteSnapshot(self.expect[0]['device'],
id = self.expect[0]['snapshots'][2]['id'],
name = self.expect[0]['snapshots'][2]['name'])
self.verifySnapshotInfo()
def test_error_device(self):
result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
device = 'drive_error',
id = '0')
self.assert_qmp(result, 'error/class', 'GenericError')
def test_error_no_id_and_name(self):
result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
device = self.expect[0]['device'])
self.assert_qmp(result, 'error/class', 'GenericError')
def test_error_snapshot_not_exist(self):
self.createSnapshotInTransaction(2)
self.verifySnapshotInfo()
result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
device = self.expect[0]['device'],
id = self.expect[0]['snapshots'][0]['id'],
name = self.expect[0]['snapshots'][1]['name'])
self.assert_qmp(result, 'error/class', 'GenericError')
if __name__ == '__main__':
iotests.main(supported_fmts=['qcow2'],
supported_protocols=['file'])
