qemu/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', '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'])