1#!/usr/bin/env python 2# 3# Tests for drive-backup 4# 5# Copyright (C) 2013 Red Hat, Inc. 6# 7# Based on 041. 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_img = os.path.join(iotests.test_dir, 'test.img') 29target_img = os.path.join(iotests.test_dir, 'target.img') 30 31class TestSingleDrive(iotests.QMPTestCase): 32 image_len = 64 * 1024 * 1024 # MB 33 34 def setUp(self): 35 # Write data to the image so we can compare later 36 qemu_img('create', '-f', iotests.imgfmt, test_img, str(TestSingleDrive.image_len)) 37 qemu_io('-c', 'write -P0x5d 0 64k', test_img) 38 qemu_io('-c', 'write -P0xd5 1M 32k', test_img) 39 qemu_io('-c', 'write -P0xdc 32M 124k', test_img) 40 qemu_io('-c', 'write -P0xdc 67043328 64k', test_img) 41 42 self.vm = iotests.VM().add_drive(test_img) 43 self.vm.launch() 44 45 def tearDown(self): 46 self.vm.shutdown() 47 os.remove(test_img) 48 try: 49 os.remove(target_img) 50 except OSError: 51 pass 52 53 def test_cancel(self): 54 self.assert_no_active_block_jobs() 55 56 result = self.vm.qmp('drive-backup', device='drive0', 57 target=target_img, sync='full') 58 self.assert_qmp(result, 'return', {}) 59 60 event = self.cancel_and_wait() 61 self.assert_qmp(event, 'data/type', 'backup') 62 63 def test_pause(self): 64 self.assert_no_active_block_jobs() 65 66 result = self.vm.qmp('drive-backup', device='drive0', 67 target=target_img, sync='full') 68 self.assert_qmp(result, 'return', {}) 69 70 result = self.vm.qmp('block-job-pause', device='drive0') 71 self.assert_qmp(result, 'return', {}) 72 73 time.sleep(1) 74 result = self.vm.qmp('query-block-jobs') 75 offset = self.dictpath(result, 'return[0]/offset') 76 77 time.sleep(1) 78 result = self.vm.qmp('query-block-jobs') 79 self.assert_qmp(result, 'return[0]/offset', offset) 80 81 result = self.vm.qmp('block-job-resume', device='drive0') 82 self.assert_qmp(result, 'return', {}) 83 84 self.wait_until_completed() 85 86 self.vm.shutdown() 87 self.assertTrue(iotests.compare_images(test_img, target_img), 88 'target image does not match source after backup') 89 90 def test_medium_not_found(self): 91 result = self.vm.qmp('drive-backup', device='ide1-cd0', 92 target=target_img, sync='full') 93 self.assert_qmp(result, 'error/class', 'GenericError') 94 95 def test_image_not_found(self): 96 result = self.vm.qmp('drive-backup', device='drive0', 97 target=target_img, sync='full', mode='existing') 98 self.assert_qmp(result, 'error/class', 'GenericError') 99 100 def test_device_not_found(self): 101 result = self.vm.qmp('drive-backup', device='nonexistent', 102 target=target_img, sync='full') 103 self.assert_qmp(result, 'error/class', 'DeviceNotFound') 104 105class TestSetSpeed(iotests.QMPTestCase): 106 image_len = 80 * 1024 * 1024 # MB 107 108 def setUp(self): 109 qemu_img('create', '-f', iotests.imgfmt, test_img, str(TestSetSpeed.image_len)) 110 self.vm = iotests.VM().add_drive(test_img) 111 self.vm.launch() 112 113 def tearDown(self): 114 self.vm.shutdown() 115 os.remove(test_img) 116 os.remove(target_img) 117 118 def test_set_speed(self): 119 self.assert_no_active_block_jobs() 120 121 result = self.vm.qmp('drive-backup', device='drive0', 122 target=target_img, sync='full') 123 self.assert_qmp(result, 'return', {}) 124 125 # Default speed is 0 126 result = self.vm.qmp('query-block-jobs') 127 self.assert_qmp(result, 'return[0]/device', 'drive0') 128 self.assert_qmp(result, 'return[0]/speed', 0) 129 130 result = self.vm.qmp('block-job-set-speed', device='drive0', speed=8 * 1024 * 1024) 131 self.assert_qmp(result, 'return', {}) 132 133 # Ensure the speed we set was accepted 134 result = self.vm.qmp('query-block-jobs') 135 self.assert_qmp(result, 'return[0]/device', 'drive0') 136 self.assert_qmp(result, 'return[0]/speed', 8 * 1024 * 1024) 137 138 event = self.cancel_and_wait() 139 self.assert_qmp(event, 'data/type', 'backup') 140 141 # Check setting speed in drive-backup works 142 result = self.vm.qmp('drive-backup', device='drive0', 143 target=target_img, sync='full', speed=4*1024*1024) 144 self.assert_qmp(result, 'return', {}) 145 146 result = self.vm.qmp('query-block-jobs') 147 self.assert_qmp(result, 'return[0]/device', 'drive0') 148 self.assert_qmp(result, 'return[0]/speed', 4 * 1024 * 1024) 149 150 event = self.cancel_and_wait() 151 self.assert_qmp(event, 'data/type', 'backup') 152 153 def test_set_speed_invalid(self): 154 self.assert_no_active_block_jobs() 155 156 result = self.vm.qmp('drive-backup', device='drive0', 157 target=target_img, sync='full', speed=-1) 158 self.assert_qmp(result, 'error/class', 'GenericError') 159 160 self.assert_no_active_block_jobs() 161 162 result = self.vm.qmp('drive-backup', device='drive0', 163 target=target_img, sync='full') 164 self.assert_qmp(result, 'return', {}) 165 166 result = self.vm.qmp('block-job-set-speed', device='drive0', speed=-1) 167 self.assert_qmp(result, 'error/class', 'GenericError') 168 169 event = self.cancel_and_wait() 170 self.assert_qmp(event, 'data/type', 'backup') 171 172class TestSingleTransaction(iotests.QMPTestCase): 173 image_len = 64 * 1024 * 1024 # MB 174 175 def setUp(self): 176 qemu_img('create', '-f', iotests.imgfmt, test_img, str(TestSingleTransaction.image_len)) 177 qemu_io('-c', 'write -P0x5d 0 64k', test_img) 178 qemu_io('-c', 'write -P0xd5 1M 32k', test_img) 179 qemu_io('-c', 'write -P0xdc 32M 124k', test_img) 180 qemu_io('-c', 'write -P0xdc 67043328 64k', test_img) 181 182 self.vm = iotests.VM().add_drive(test_img) 183 self.vm.launch() 184 185 def tearDown(self): 186 self.vm.shutdown() 187 os.remove(test_img) 188 try: 189 os.remove(target_img) 190 except OSError: 191 pass 192 193 def test_cancel(self): 194 self.assert_no_active_block_jobs() 195 196 result = self.vm.qmp('transaction', actions=[{ 197 'type': 'drive-backup', 198 'data': { 'device': 'drive0', 199 'target': target_img, 200 'sync': 'full' }, 201 } 202 ]) 203 self.assert_qmp(result, 'return', {}) 204 205 event = self.cancel_and_wait() 206 self.assert_qmp(event, 'data/type', 'backup') 207 208 def test_pause(self): 209 self.assert_no_active_block_jobs() 210 211 result = self.vm.qmp('transaction', actions=[{ 212 'type': 'drive-backup', 213 'data': { 'device': 'drive0', 214 'target': target_img, 215 'sync': 'full' }, 216 } 217 ]) 218 self.assert_qmp(result, 'return', {}) 219 220 result = self.vm.qmp('block-job-pause', device='drive0') 221 self.assert_qmp(result, 'return', {}) 222 223 time.sleep(1) 224 result = self.vm.qmp('query-block-jobs') 225 offset = self.dictpath(result, 'return[0]/offset') 226 227 time.sleep(1) 228 result = self.vm.qmp('query-block-jobs') 229 self.assert_qmp(result, 'return[0]/offset', offset) 230 231 result = self.vm.qmp('block-job-resume', device='drive0') 232 self.assert_qmp(result, 'return', {}) 233 234 self.wait_until_completed() 235 236 self.vm.shutdown() 237 self.assertTrue(iotests.compare_images(test_img, target_img), 238 'target image does not match source after backup') 239 240 def test_medium_not_found(self): 241 result = self.vm.qmp('transaction', actions=[{ 242 'type': 'drive-backup', 243 'data': { 'device': 'ide1-cd0', 244 'target': target_img, 245 'sync': 'full' }, 246 } 247 ]) 248 self.assert_qmp(result, 'error/class', 'GenericError') 249 250 def test_image_not_found(self): 251 result = self.vm.qmp('transaction', actions=[{ 252 'type': 'drive-backup', 253 'data': { 'device': 'drive0', 254 'mode': 'existing', 255 'target': target_img, 256 'sync': 'full' }, 257 } 258 ]) 259 self.assert_qmp(result, 'error/class', 'GenericError') 260 261 def test_device_not_found(self): 262 result = self.vm.qmp('transaction', actions=[{ 263 'type': 'drive-backup', 264 'data': { 'device': 'nonexistent', 265 'mode': 'existing', 266 'target': target_img, 267 'sync': 'full' }, 268 } 269 ]) 270 self.assert_qmp(result, 'error/class', 'DeviceNotFound') 271 272 def test_abort(self): 273 result = self.vm.qmp('transaction', actions=[{ 274 'type': 'drive-backup', 275 'data': { 'device': 'nonexistent', 276 'mode': 'existing', 277 'target': target_img, 278 'sync': 'full' }, 279 }, { 280 'type': 'Abort', 281 'data': {}, 282 } 283 ]) 284 self.assert_qmp(result, 'error/class', 'GenericError') 285 self.assert_no_active_block_jobs() 286 287if __name__ == '__main__': 288 iotests.main(supported_fmts=['raw', 'qcow2']) 289