1#!/usr/bin/env python3 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 23import time 24import os 25import iotests 26from iotests import qemu_img, qemu_io 27 28test_drv_base_name = 'drive' 29 30class 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 163class 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 197class 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 211class 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 258if __name__ == '__main__': 259 iotests.main(supported_fmts=['qcow2'], 260 supported_protocols=['file']) 261