[qemu.git] / tests / qemu-iotests / 057

#!/usr/bin/env python
#
# 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', 'DeviceNotFound')

    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', 'DeviceNotFound')

    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'])
Commit	Line	Data
8023090b WX	1	#!/usr/bin/env python
	2	#
	3	# Tests for internal snapshot.
	4	#
	5	# Copyright (C) 2013 IBM, Inc.
	6	#
	7	# Based on 055.
	8	#
	9	# This program is free software; you can redistribute it and/or modify
	10	# it under the terms of the GNU General Public License as published by
	11	# the Free Software Foundation; either version 2 of the License, or
	12	# (at your option) any later version.
	13	#
	14	# This program is distributed in the hope that it will be useful,
	15	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	16	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	17	# GNU General Public License for more details.
	18	#
	19	# You should have received a copy of the GNU General Public License
	20	# along with this program. If not, see <http://www.gnu.org/licenses/>.
	21	#
	22
	23	import time
	24	import os
	25	import iotests
	26	from iotests import qemu_img, qemu_io
	27
	28	test_drv_base_name = 'drive'
	29
	30	class ImageSnapshotTestCase(iotests.QMPTestCase):
	31	image_len = 120 * 1024 * 1024 # MB
	32
	33	def __init__(self, *args):
	34	self.expect = []
	35	super(ImageSnapshotTestCase, self).__init__(*args)
	36
	37	def _setUp(self, test_img_base_name, image_num):
	38	self.vm = iotests.VM()
	39	for i in range(0, image_num):
	40	filename = '%s%d' % (test_img_base_name, i)
	41	img = os.path.join(iotests.test_dir, filename)
	42	device = '%s%d' % (test_drv_base_name, i)
	43	qemu_img('create', '-f', iotests.imgfmt, img, str(self.image_len))
	44	self.vm.add_drive(img)
	45	self.expect.append({'image': img, 'device': device,
	46	'snapshots': [],
	47	'snapshots_name_counter': 0})
	48	self.vm.launch()
	49
	50	def tearDown(self):
	51	self.vm.shutdown()
	52	for dev_expect in self.expect:
	53	os.remove(dev_expect['image'])
	54
	55	def createSnapshotInTransaction(self, snapshot_num, abort = False):
	56	actions = []
	57	for dev_expect in self.expect:
	58	num = dev_expect['snapshots_name_counter']
	59	for j in range(0, snapshot_num):
	60	name = '%s_sn%d' % (dev_expect['device'], num)
	61	num = num + 1
	62	if abort == False:
	63	dev_expect['snapshots'].append({'name': name})
	64	dev_expect['snapshots_name_counter'] = num
65	actions.append({
66	'type': 'blockdev-snapshot-internal-sync',
67	'data': { 'device': dev_expect['device'],
68	'name': name },
69	})
70
71	if abort == True:
72	actions.append({
73	'type': 'abort',
74	'data': {},
75	})
76
77	result = self.vm.qmp('transaction', actions = actions)
78
79	if abort == True:
80	self.assert_qmp(result, 'error/class', 'GenericError')
81	else:
82	self.assert_qmp(result, 'return', {})
83
84	def verifySnapshotInfo(self):
85	result = self.vm.qmp('query-block')
86
87	# Verify each expected result
88	for dev_expect in self.expect:
89	# 1. Find the returned image value and snapshot info
90	image_result = None
91	for device in result['return']:
92	if device['device'] == dev_expect['device']:
93	image_result = device['inserted']['image']
94	break
95	self.assertTrue(image_result != None)
96	# Do not consider zero snapshot case now
97	sn_list_result = image_result['snapshots']
98	sn_list_expect = dev_expect['snapshots']
99
100	# 2. Verify it with expect
101	self.assertTrue(len(sn_list_result) == len(sn_list_expect))
102
103	for sn_expect in sn_list_expect:
104	sn_result = None
105	for sn in sn_list_result:
106	if sn_expect['name'] == sn['name']:
107	sn_result = sn
108	break
109	self.assertTrue(sn_result != None)
110	# Fill in the detail info
111	sn_expect.update(sn_result)
112
113	def deleteSnapshot(self, device, id = None, name = None):
114	sn_list_expect = None
115	sn_expect = None
116
117	self.assertTrue(id != None or name != None)
118
119	# Fill in the detail info include ID
120	self.verifySnapshotInfo()
121
122	#find the expected snapshot list
123	for dev_expect in self.expect:
124	if dev_expect['device'] == device:
125	sn_list_expect = dev_expect['snapshots']
126	break
127	self.assertTrue(sn_list_expect != None)
128
129	if id != None and name != None:
130	for sn in sn_list_expect:
131	if sn['id'] == id and sn['name'] == name:
132	sn_expect = sn
133	result = \
134	self.vm.qmp('blockdev-snapshot-delete-internal-sync',
135	device = device,
136	id = id,
137	name = name)
138	break
139	elif id != None:
140	for sn in sn_list_expect:
141	if sn['id'] == id:
142	sn_expect = sn
143	result = \
144	self.vm.qmp('blockdev-snapshot-delete-internal-sync',
145	device = device,
146	id = id)
147	break
148	else:
149	for sn in sn_list_expect:
150	if sn['name'] == name:
151	sn_expect = sn
152	result = \
153	self.vm.qmp('blockdev-snapshot-delete-internal-sync',
154	device = device,
155	name = name)
156	break
157
158	self.assertTrue(sn_expect != None)
159
160	self.assert_qmp(result, 'return', sn_expect)
161	sn_list_expect.remove(sn_expect)
162
163	class TestSingleTransaction(ImageSnapshotTestCase):
164	def setUp(self):
165	self._setUp('test_a.img', 1)
166
167	def test_create(self):
168	self.createSnapshotInTransaction(1)
169	self.verifySnapshotInfo()
170
171	def test_error_name_empty(self):
172	actions = [{'type': 'blockdev-snapshot-internal-sync',
173	'data': { 'device': self.expect[0]['device'],
174	'name': '' },
175	}]
176	result = self.vm.qmp('transaction', actions = actions)
177	self.assert_qmp(result, 'error/class', 'GenericError')
178
179	def test_error_device(self):
180	actions = [{'type': 'blockdev-snapshot-internal-sync',
181	'data': { 'device': 'drive_error',
182	'name': 'a' },
183	}]
184	result = self.vm.qmp('transaction', actions = actions)
185	self.assert_qmp(result, 'error/class', 'DeviceNotFound')
186
187	def test_error_exist(self):
188	self.createSnapshotInTransaction(1)
189	self.verifySnapshotInfo()
190	actions = [{'type': 'blockdev-snapshot-internal-sync',
191	'data': { 'device': self.expect[0]['device'],
192	'name': self.expect[0]['snapshots'][0] },
193	}]
194	result = self.vm.qmp('transaction', actions = actions)
195	self.assert_qmp(result, 'error/class', 'GenericError')
196
197	class TestMultipleTransaction(ImageSnapshotTestCase):
198	def setUp(self):
199	self._setUp('test_b.img', 2)
200
201	def test_create(self):
202	self.createSnapshotInTransaction(3)
203	self.verifySnapshotInfo()
204
205	def test_abort(self):
206	self.createSnapshotInTransaction(2)
207	self.verifySnapshotInfo()
208	self.createSnapshotInTransaction(3, abort = True)
209	self.verifySnapshotInfo()
210
211	class TestSnapshotDelete(ImageSnapshotTestCase):
212	def setUp(self):
213	self._setUp('test_c.img', 1)
214
215	def test_delete_with_id(self):
216	self.createSnapshotInTransaction(2)
217	self.verifySnapshotInfo()
218	self.deleteSnapshot(self.expect[0]['device'],
219	id = self.expect[0]['snapshots'][0]['id'])
220	self.verifySnapshotInfo()
221
222	def test_delete_with_name(self):
223	self.createSnapshotInTransaction(3)
224	self.verifySnapshotInfo()
225	self.deleteSnapshot(self.expect[0]['device'],
226	name = self.expect[0]['snapshots'][1]['name'])
227	self.verifySnapshotInfo()
228
229	def test_delete_with_id_and_name(self):
230	self.createSnapshotInTransaction(4)
231	self.verifySnapshotInfo()
232	self.deleteSnapshot(self.expect[0]['device'],
233	id = self.expect[0]['snapshots'][2]['id'],
234	name = self.expect[0]['snapshots'][2]['name'])
235	self.verifySnapshotInfo()
236
237
238	def test_error_device(self):
239	result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
240	device = 'drive_error',
241	id = '0')
242	self.assert_qmp(result, 'error/class', 'DeviceNotFound')
243
244	def test_error_no_id_and_name(self):
245	result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
246	device = self.expect[0]['device'])
247	self.assert_qmp(result, 'error/class', 'GenericError')
248
249	def test_error_snapshot_not_exist(self):
250	self.createSnapshotInTransaction(2)
251	self.verifySnapshotInfo()
252	result = self.vm.qmp('blockdev-snapshot-delete-internal-sync',
253	device = self.expect[0]['device'],
254	id = self.expect[0]['snapshots'][0]['id'],
255	name = self.expect[0]['snapshots'][1]['name'])
256	self.assert_qmp(result, 'error/class', 'GenericError')
257
258	if __name__ == '__main__':
259	iotests.main(supported_fmts=['qcow2'])