| #!/usr/bin/env python3 |
| # |
| # Tests for internal snapshot. |
| # |
| # Copyright (C) 2013 IBM, Inc. |
| # |
| # Based on 055. |
| # |
| # This program is free software; you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License as published by |
| # the Free Software Foundation; either version 2 of the License, or |
| # (at your option) any later version. |
| # |
| # This program is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| # GNU General Public License for more details. |
| # |
| # You should have received a copy of the GNU General Public License |
| # along with this program. If not, see <http://www.gnu.org/licenses/>. |
| # |
| |
| import time |
| import os |
| import iotests |
| from iotests import qemu_img, qemu_io |
| |
| test_drv_base_name = 'drive' |
| |
| class ImageSnapshotTestCase(iotests.QMPTestCase): |
| image_len = 120 * 1024 * 1024 # MB |
| |
| def __init__(self, *args): |
| self.expect = [] |
| super(ImageSnapshotTestCase, self).__init__(*args) |
| |
| def _setUp(self, test_img_base_name, image_num): |
| self.vm = iotests.VM() |
| for i in range(0, image_num): |
| filename = '%s%d' % (test_img_base_name, i) |
| img = os.path.join(iotests.test_dir, filename) |
| device = '%s%d' % (test_drv_base_name, i) |
| qemu_img('create', '-f', iotests.imgfmt, img, str(self.image_len)) |
| self.vm.add_drive(img) |
| self.expect.append({'image': img, 'device': device, |
| 'snapshots': [], |
| 'snapshots_name_counter': 0}) |
| self.vm.launch() |
| |
| def tearDown(self): |
| self.vm.shutdown() |
| for dev_expect in self.expect: |
| os.remove(dev_expect['image']) |
| |
| def createSnapshotInTransaction(self, snapshot_num, abort = False): |
| actions = [] |
| for dev_expect in self.expect: |
| num = dev_expect['snapshots_name_counter'] |
| for j in range(0, snapshot_num): |
| name = '%s_sn%d' % (dev_expect['device'], num) |
| num = num + 1 |
| if abort == False: |
| dev_expect['snapshots'].append({'name': name}) |
| dev_expect['snapshots_name_counter'] = num |
| actions.append({ |
| 'type': 'blockdev-snapshot-internal-sync', |
| 'data': { 'device': dev_expect['device'], |
| 'name': name }, |
| }) |
| |
| if abort == True: |
| actions.append({ |
| 'type': 'abort', |
| 'data': {}, |
| }) |
| |
| result = self.vm.qmp('transaction', actions = actions) |
| |
| if abort == True: |
| self.assert_qmp(result, 'error/class', 'GenericError') |
| else: |
| self.assert_qmp(result, 'return', {}) |
| |
| def verifySnapshotInfo(self): |
| result = self.vm.qmp('query-block') |
| |
| # Verify each expected result |
| for dev_expect in self.expect: |
| # 1. Find the returned image value and snapshot info |
| image_result = None |
| for device in result['return']: |
| if device['device'] == dev_expect['device']: |
| image_result = device['inserted']['image'] |
| break |
| self.assertTrue(image_result != None) |
| # Do not consider zero snapshot case now |
| sn_list_result = image_result['snapshots'] |
| sn_list_expect = dev_expect['snapshots'] |
| |
| # 2. Verify it with expect |
| self.assertTrue(len(sn_list_result) == len(sn_list_expect)) |
| |
| for sn_expect in sn_list_expect: |
| sn_result = None |
| for sn in sn_list_result: |
| if sn_expect['name'] == sn['name']: |
| sn_result = sn |
| break |
| self.assertTrue(sn_result != None) |
| # Fill in the detail info |
| sn_expect.update(sn_result) |
| |
| def deleteSnapshot(self, device, id = None, name = None): |
| sn_list_expect = None |
| sn_expect = None |
| |
| self.assertTrue(id != None or name != None) |
| |
| # Fill in the detail info include ID |
| self.verifySnapshotInfo() |
| |
| #find the expected snapshot list |
| for dev_expect in self.expect: |
| if dev_expect['device'] == device: |
| sn_list_expect = dev_expect['snapshots'] |
| break |
| self.assertTrue(sn_list_expect != None) |
| |
| if id != None and name != None: |
| for sn in sn_list_expect: |
| if sn['id'] == id and sn['name'] == name: |
| sn_expect = sn |
| result = \ |
| self.vm.qmp('blockdev-snapshot-delete-internal-sync', |
| device = device, |
| id = id, |
| name = name) |
| break |
| elif id != None: |
| for sn in sn_list_expect: |
| if sn['id'] == id: |
| sn_expect = sn |
| result = \ |
| self.vm.qmp('blockdev-snapshot-delete-internal-sync', |
| device = device, |
| id = id) |
| break |
| else: |
| for sn in sn_list_expect: |
| if sn['name'] == name: |
| sn_expect = sn |
| result = \ |
| self.vm.qmp('blockdev-snapshot-delete-internal-sync', |
| device = device, |
| name = name) |
| break |
| |
| self.assertTrue(sn_expect != None) |
| |
| self.assert_qmp(result, 'return', sn_expect) |
| sn_list_expect.remove(sn_expect) |
| |
| class TestSingleTransaction(ImageSnapshotTestCase): |
| def setUp(self): |
| self._setUp('test_a.img', 1) |
| |
| def test_create(self): |
| self.createSnapshotInTransaction(1) |
| self.verifySnapshotInfo() |
| |
| def test_error_name_empty(self): |
| actions = [{'type': 'blockdev-snapshot-internal-sync', |
| 'data': { 'device': self.expect[0]['device'], |
| 'name': '' }, |
| }] |
| result = self.vm.qmp('transaction', actions = actions) |
| self.assert_qmp(result, 'error/class', 'GenericError') |
| |
| def test_error_device(self): |
| actions = [{'type': 'blockdev-snapshot-internal-sync', |
| 'data': { 'device': 'drive_error', |
| 'name': 'a' }, |
| }] |
| result = self.vm.qmp('transaction', actions = actions) |
| self.assert_qmp(result, 'error/class', 'GenericError') |
| |
| def test_error_exist(self): |
| self.createSnapshotInTransaction(1) |
| self.verifySnapshotInfo() |
| actions = [{'type': 'blockdev-snapshot-internal-sync', |
| 'data': { 'device': self.expect[0]['device'], |
| 'name': self.expect[0]['snapshots'][0] }, |
| }] |
| result = self.vm.qmp('transaction', actions = actions) |
| self.assert_qmp(result, 'error/class', 'GenericError') |
| |
| class TestMultipleTransaction(ImageSnapshotTestCase): |
| def setUp(self): |
| self._setUp('test_b.img', 2) |
| |
| def test_create(self): |
| self.createSnapshotInTransaction(3) |
| self.verifySnapshotInfo() |
| |
| def test_abort(self): |
| self.createSnapshotInTransaction(2) |
| self.verifySnapshotInfo() |
| self.createSnapshotInTransaction(3, abort = True) |
| self.verifySnapshotInfo() |
| |
| class TestSnapshotDelete(ImageSnapshotTestCase): |
| def setUp(self): |
| self._setUp('test_c.img', 1) |
| |
| def test_delete_with_id(self): |
| self.createSnapshotInTransaction(2) |
| self.verifySnapshotInfo() |
| self.deleteSnapshot(self.expect[0]['device'], |
| id = self.expect[0]['snapshots'][0]['id']) |
| self.verifySnapshotInfo() |
| |
| def test_delete_with_name(self): |
| self.createSnapshotInTransaction(3) |
| self.verifySnapshotInfo() |
| self.deleteSnapshot(self.expect[0]['device'], |
| name = self.expect[0]['snapshots'][1]['name']) |
| self.verifySnapshotInfo() |
| |
| def test_delete_with_id_and_name(self): |
| self.createSnapshotInTransaction(4) |
| self.verifySnapshotInfo() |
| self.deleteSnapshot(self.expect[0]['device'], |
| id = self.expect[0]['snapshots'][2]['id'], |
| name = self.expect[0]['snapshots'][2]['name']) |
| self.verifySnapshotInfo() |
| |
| |
| def test_error_device(self): |
| result = self.vm.qmp('blockdev-snapshot-delete-internal-sync', |
| device = 'drive_error', |
| id = '0') |
| self.assert_qmp(result, 'error/class', 'GenericError') |
| |
| def test_error_no_id_and_name(self): |
| result = self.vm.qmp('blockdev-snapshot-delete-internal-sync', |
| device = self.expect[0]['device']) |
| self.assert_qmp(result, 'error/class', 'GenericError') |
| |
| def test_error_snapshot_not_exist(self): |
| self.createSnapshotInTransaction(2) |
| self.verifySnapshotInfo() |
| result = self.vm.qmp('blockdev-snapshot-delete-internal-sync', |
| device = self.expect[0]['device'], |
| id = self.expect[0]['snapshots'][0]['id'], |
| name = self.expect[0]['snapshots'][1]['name']) |
| self.assert_qmp(result, 'error/class', 'GenericError') |
| |
| if __name__ == '__main__': |
| iotests.main(supported_fmts=['qcow2'], |
| supported_protocols=['file']) |
