cirrus: fix off-by-one in cirrus_bitblt_rop_bkwd_transp_*_16
[qemu.git] / tests / qemu-iotests / 057
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', 'GenericError')
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', 'GenericError')
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'])