tests/qemu-iotests/055

   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
  23 import time
  24 import os
  25 import iotests
  26 from iotests import qemu_img, qemu_io
  27
  28 test_img = os.path.join(iotests.test_dir, 'test.img')
  29 target_img = os.path.join(iotests.test_dir, 'target.img')
  30
  31 class 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)
  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)
  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)
  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                              mode='existing', target=target_img)
  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)
 103         self.assert_qmp(result, 'error/class', 'DeviceNotFound')
 104
 105 class 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)
 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, 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, 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)
 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
 172 class 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             }
 201         ])
 202         self.assert_qmp(result, 'return', {})
 203
 204         event = self.cancel_and_wait()
 205         self.assert_qmp(event, 'data/type', 'backup')
 206
 207     def test_pause(self):
 208         self.assert_no_active_block_jobs()
 209
 210         result = self.vm.qmp('transaction', actions=[{
 211                 'type': 'drive-backup',
 212                 'data': { 'device': 'drive0',
 213                           'target': target_img },
 214             }
 215         ])
 216         self.assert_qmp(result, 'return', {})
 217
 218         result = self.vm.qmp('block-job-pause', device='drive0')
 219         self.assert_qmp(result, 'return', {})
 220
 221         time.sleep(1)
 222         result = self.vm.qmp('query-block-jobs')
 223         offset = self.dictpath(result, 'return[0]/offset')
 224
 225         time.sleep(1)
 226         result = self.vm.qmp('query-block-jobs')
 227         self.assert_qmp(result, 'return[0]/offset', offset)
 228
 229         result = self.vm.qmp('block-job-resume', device='drive0')
 230         self.assert_qmp(result, 'return', {})
 231
 232         self.wait_until_completed()
 233
 234         self.vm.shutdown()
 235         self.assertTrue(iotests.compare_images(test_img, target_img),
 236                         'target image does not match source after backup')
 237
 238     def test_medium_not_found(self):
 239         result = self.vm.qmp('transaction', actions=[{
 240                 'type': 'drive-backup',
 241                 'data': { 'device': 'ide1-cd0',
 242                           'target': target_img },
 243             }
 244         ])
 245         self.assert_qmp(result, 'error/class', 'GenericError')
 246
 247     def test_image_not_found(self):
 248         result = self.vm.qmp('transaction', actions=[{
 249                 'type': 'drive-backup',
 250                 'data': { 'device': 'drive0',
 251                           'mode': 'existing',
 252                           'target': target_img },
 253             }
 254         ])
 255         self.assert_qmp(result, 'error/class', 'GenericError')
 256
 257     def test_device_not_found(self):
 258         result = self.vm.qmp('transaction', actions=[{
 259                 'type': 'drive-backup',
 260                 'data': { 'device': 'nonexistent',
 261                           'mode': 'existing',
 262                           'target': target_img },
 263             }
 264         ])
 265         self.assert_qmp(result, 'error/class', 'DeviceNotFound')
 266
 267     def test_abort(self):
 268         result = self.vm.qmp('transaction', actions=[{
 269                 'type': 'drive-backup',
 270                 'data': { 'device': 'nonexistent',
 271                           'mode': 'existing',
 272                           'target': target_img },
 273             }, {
 274                 'type': 'Abort',
 275                 'data': {},
 276             }
 277         ])
 278         self.assert_qmp(result, 'error/class', 'GenericError')
 279         self.assert_no_active_block_jobs()
 280
 281 if __name__ == '__main__':
 282     iotests.main(supported_fmts=['raw', 'qcow2'])