[qemu.git] / tests / qemu-iotests / 139

#!/usr/bin/env python
#
# Test cases for the QMP 'x-blockdev-del' command
#
# Copyright (C) 2015 Igalia, S.L.
# Author: Alberto Garcia <[email protected]>
#
# 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 os
import iotests
import time

base_img = os.path.join(iotests.test_dir, 'base.img')
new_img = os.path.join(iotests.test_dir, 'new.img')

class TestBlockdevDel(iotests.QMPTestCase):

    def setUp(self):
        iotests.qemu_img('create', '-f', iotests.imgfmt, base_img, '1M')
        self.vm = iotests.VM()
        self.vm.launch()

    def tearDown(self):
        self.vm.shutdown()
        os.remove(base_img)
        if os.path.isfile(new_img):
            os.remove(new_img)

    # Check whether a BlockBackend exists
    def checkBlockBackend(self, backend, node, must_exist = True):
        result = self.vm.qmp('query-block')
        backends = filter(lambda x: x['device'] == backend, result['return'])
        self.assertLessEqual(len(backends), 1)
        self.assertEqual(must_exist, len(backends) == 1)
        if must_exist:
            if node:
                self.assertEqual(backends[0]['inserted']['node-name'], node)
            else:
                self.assertFalse(backends[0].has_key('inserted'))

    # Check whether a BlockDriverState exists
    def checkBlockDriverState(self, node, must_exist = True):
        result = self.vm.qmp('query-named-block-nodes')
        nodes = filter(lambda x: x['node-name'] == node, result['return'])
        self.assertLessEqual(len(nodes), 1)
        self.assertEqual(must_exist, len(nodes) == 1)

    # Add a new BlockBackend (with its attached BlockDriverState)
    def addBlockBackend(self, backend, node):
        file_node = '%s_file' % node
        self.checkBlockBackend(backend, node, False)
        self.checkBlockDriverState(node, False)
        self.checkBlockDriverState(file_node, False)
        opts = {'driver': iotests.imgfmt,
                'id': backend,
                'node-name': node,
                'file': {'driver': 'file',
                         'node-name': file_node,
                         'filename': base_img}}
        result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockBackend(backend, node)
        self.checkBlockDriverState(node)
        self.checkBlockDriverState(file_node)

    # Add a BlockDriverState without a BlockBackend
    def addBlockDriverState(self, node):
        file_node = '%s_file' % node
        self.checkBlockDriverState(node, False)
        self.checkBlockDriverState(file_node, False)
        opts = {'driver': iotests.imgfmt,
                'node-name': node,
                'file': {'driver': 'file',
                         'node-name': file_node,
                         'filename': base_img}}
        result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(node)
        self.checkBlockDriverState(file_node)

    # Add a BlockDriverState that will be used as overlay for the base_img BDS
    def addBlockDriverStateOverlay(self, node):
        self.checkBlockDriverState(node, False)
        iotests.qemu_img('create', '-f', iotests.imgfmt,
                         '-b', base_img, new_img, '1M')
        opts = {'driver': iotests.imgfmt,
                'node-name': node,
                'backing': '',
                'file': {'driver': 'file',
                         'filename': new_img}}
        result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(node)

    # Delete a BlockBackend
    def delBlockBackend(self, backend, node, expect_error = False,
                        destroys_media = True):
        self.checkBlockBackend(backend, node)
        if node:
            self.checkBlockDriverState(node)
        result = self.vm.qmp('x-blockdev-del', id = backend)
        if expect_error:
            self.assert_qmp(result, 'error/class', 'GenericError')
            if node:
                self.checkBlockDriverState(node)
        else:
            self.assert_qmp(result, 'return', {})
            if node:
                self.checkBlockDriverState(node, not destroys_media)
        self.checkBlockBackend(backend, node, must_exist = expect_error)

    # Delete a BlockDriverState
    def delBlockDriverState(self, node, expect_error = False):
        self.checkBlockDriverState(node)
        result = self.vm.qmp('x-blockdev-del', node_name = node)
        if expect_error:
            self.assert_qmp(result, 'error/class', 'GenericError')
        else:
            self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(node, expect_error)

    # Add a device model
    def addDeviceModel(self, device, backend):
        result = self.vm.qmp('device_add', id = device,
                             driver = 'virtio-blk-pci', drive = backend)
        self.assert_qmp(result, 'return', {})

    # Delete a device model
    def delDeviceModel(self, device):
        result = self.vm.qmp('device_del', id = device)
        self.assert_qmp(result, 'return', {})

        result = self.vm.qmp('system_reset')
        self.assert_qmp(result, 'return', {})

        device_path = '/machine/peripheral/%s/virtio-backend' % device
        event = self.vm.event_wait(name="DEVICE_DELETED",
                                   match={'data': {'path': device_path}})
        self.assertNotEqual(event, None)

        event = self.vm.event_wait(name="DEVICE_DELETED",
                                   match={'data': {'device': device}})
        self.assertNotEqual(event, None)

    # Remove a BlockDriverState
    def ejectDrive(self, backend, node, expect_error = False,
                   destroys_media = True):
        self.checkBlockBackend(backend, node)
        self.checkBlockDriverState(node)
        result = self.vm.qmp('eject', device = backend)
        if expect_error:
            self.assert_qmp(result, 'error/class', 'GenericError')
            self.checkBlockDriverState(node)
            self.checkBlockBackend(backend, node)
        else:
            self.assert_qmp(result, 'return', {})
            self.checkBlockDriverState(node, not destroys_media)
            self.checkBlockBackend(backend, None)

    # Insert a BlockDriverState
    def insertDrive(self, backend, node):
        self.checkBlockBackend(backend, None)
        self.checkBlockDriverState(node)
        result = self.vm.qmp('x-blockdev-insert-medium',
                             device = backend, node_name = node)
        self.assert_qmp(result, 'return', {})
        self.checkBlockBackend(backend, node)
        self.checkBlockDriverState(node)

    # Create a snapshot using 'blockdev-snapshot-sync'
    def createSnapshotSync(self, node, overlay):
        self.checkBlockDriverState(node)
        self.checkBlockDriverState(overlay, False)
        opts = {'node-name': node,
                'snapshot-file': new_img,
                'snapshot-node-name': overlay,
                'format': iotests.imgfmt}
        result = self.vm.qmp('blockdev-snapshot-sync', conv_keys=False, **opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(node)
        self.checkBlockDriverState(overlay)

    # Create a snapshot using 'blockdev-snapshot'
    def createSnapshot(self, node, overlay):
        self.checkBlockDriverState(node)
        self.checkBlockDriverState(overlay)
        result = self.vm.qmp('blockdev-snapshot',
                             node = node, overlay = overlay)
        self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(node)
        self.checkBlockDriverState(overlay)

    # Create a mirror
    def createMirror(self, backend, node, new_node):
        self.checkBlockBackend(backend, node)
        self.checkBlockDriverState(new_node, False)
        opts = {'device': backend,
                'target': new_img,
                'node-name': new_node,
                'sync': 'top',
                'format': iotests.imgfmt}
        result = self.vm.qmp('drive-mirror', conv_keys=False, **opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockBackend(backend, node)
        self.checkBlockDriverState(new_node)

    # Complete an existing block job
    def completeBlockJob(self, backend, node_before, node_after):
        self.checkBlockBackend(backend, node_before)
        result = self.vm.qmp('block-job-complete', device=backend)
        self.assert_qmp(result, 'return', {})
        self.wait_until_completed(backend)
        self.checkBlockBackend(backend, node_after)

    # Add a BlkDebug node
    # Note that the purpose of this is to test the x-blockdev-del
    # sanity checks, not to create a usable blkdebug drive
    def addBlkDebug(self, debug, node):
        self.checkBlockDriverState(node, False)
        self.checkBlockDriverState(debug, False)
        image = {'driver': iotests.imgfmt,
                 'node-name': node,
                 'file': {'driver': 'file',
                          'filename': base_img}}
        opts = {'driver': 'blkdebug',
                'node-name': debug,
                'image': image}
        result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(node)
        self.checkBlockDriverState(debug)

    # Add a BlkVerify node
    # Note that the purpose of this is to test the x-blockdev-del
    # sanity checks, not to create a usable blkverify drive
    def addBlkVerify(self, blkverify, test, raw):
        self.checkBlockDriverState(test, False)
        self.checkBlockDriverState(raw, False)
        self.checkBlockDriverState(blkverify, False)
        iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M')
        node_0 = {'driver': iotests.imgfmt,
                  'node-name': test,
                  'file': {'driver': 'file',
                           'filename': base_img}}
        node_1 = {'driver': iotests.imgfmt,
                  'node-name': raw,
                  'file': {'driver': 'file',
                           'filename': new_img}}
        opts = {'driver': 'blkverify',
                'node-name': blkverify,
                'test': node_0,
                'raw': node_1}
        result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(test)
        self.checkBlockDriverState(raw)
        self.checkBlockDriverState(blkverify)

    # Add a Quorum node
    def addQuorum(self, quorum, child0, child1):
        self.checkBlockDriverState(child0, False)
        self.checkBlockDriverState(child1, False)
        self.checkBlockDriverState(quorum, False)
        iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M')
        child_0 = {'driver': iotests.imgfmt,
                   'node-name': child0,
                   'file': {'driver': 'file',
                            'filename': base_img}}
        child_1 = {'driver': iotests.imgfmt,
                   'node-name': child1,
                   'file': {'driver': 'file',
                            'filename': new_img}}
        opts = {'driver': 'quorum',
                'node-name': quorum,
                'vote-threshold': 1,
                'children': [ child_0, child_1 ]}
        result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
        self.assert_qmp(result, 'return', {})
        self.checkBlockDriverState(child0)
        self.checkBlockDriverState(child1)
        self.checkBlockDriverState(quorum)

    ########################
    # The tests start here #
    ########################

    def testWrongParameters(self):
        self.addBlockBackend('drive0', 'node0')
        result = self.vm.qmp('x-blockdev-del')
        self.assert_qmp(result, 'error/class', 'GenericError')
        result = self.vm.qmp('x-blockdev-del', id='drive0', node_name='node0')
        self.assert_qmp(result, 'error/class', 'GenericError')
        self.delBlockBackend('drive0', 'node0')

    def testBlockBackend(self):
        self.addBlockBackend('drive0', 'node0')
        # You cannot delete a BDS that is attached to a backend
        self.delBlockDriverState('node0', expect_error = True)
        self.delBlockBackend('drive0', 'node0')

    def testBlockDriverState(self):
        self.addBlockDriverState('node0')
        # You cannot delete a file BDS directly
        self.delBlockDriverState('node0_file', expect_error = True)
        self.delBlockDriverState('node0')

    def testEject(self):
        self.addBlockBackend('drive0', 'node0')
        self.ejectDrive('drive0', 'node0')
        self.delBlockBackend('drive0', None)

    def testDeviceModel(self):
        self.addBlockBackend('drive0', 'node0')
        self.addDeviceModel('device0', 'drive0')
        self.ejectDrive('drive0', 'node0', expect_error = True)
        self.delBlockBackend('drive0', 'node0', expect_error = True)
        self.delDeviceModel('device0')
        self.delBlockBackend('drive0', 'node0')

    def testAttachMedia(self):
        # This creates a BlockBackend and removes its media
        self.addBlockBackend('drive0', 'node0')
        self.ejectDrive('drive0', 'node0')
        # This creates a new BlockDriverState and inserts it into the backend
        self.addBlockDriverState('node1')
        self.insertDrive('drive0', 'node1')
        # The backend can't be removed: the new BDS has an extra reference
        self.delBlockBackend('drive0', 'node1', expect_error = True)
        self.delBlockDriverState('node1', expect_error = True)
        # The BDS still exists after being ejected, but now it can be removed
        self.ejectDrive('drive0', 'node1', destroys_media = False)
        self.delBlockDriverState('node1')
        self.delBlockBackend('drive0', None)

    def testSnapshotSync(self):
        self.addBlockBackend('drive0', 'node0')
        self.createSnapshotSync('node0', 'overlay0')
        # This fails because node0 is now being used as a backing image
        self.delBlockDriverState('node0', expect_error = True)
        # This succeeds because overlay0 only has the backend reference
        self.delBlockBackend('drive0', 'overlay0')
        self.checkBlockDriverState('node0', False)

    def testSnapshot(self):
        self.addBlockBackend('drive0', 'node0')
        self.addBlockDriverStateOverlay('overlay0')
        self.createSnapshot('node0', 'overlay0')
        self.delBlockBackend('drive0', 'overlay0', expect_error = True)
        self.delBlockDriverState('node0', expect_error = True)
        self.delBlockDriverState('overlay0', expect_error = True)
        self.ejectDrive('drive0', 'overlay0', destroys_media = False)
        self.delBlockBackend('drive0', None)
        self.delBlockDriverState('node0', expect_error = True)
        self.delBlockDriverState('overlay0')
        self.checkBlockDriverState('node0', False)

    def testMirror(self):
        self.addBlockBackend('drive0', 'node0')
        self.createMirror('drive0', 'node0', 'mirror0')
        # The block job prevents removing the device
        self.delBlockBackend('drive0', 'node0', expect_error = True)
        self.delBlockDriverState('node0', expect_error = True)
        self.delBlockDriverState('mirror0', expect_error = True)
        self.wait_ready('drive0')
        self.completeBlockJob('drive0', 'node0', 'mirror0')
        self.assert_no_active_block_jobs()
        self.checkBlockDriverState('node0', False)
        # This succeeds because the backend now points to mirror0
        self.delBlockBackend('drive0', 'mirror0')

    def testBlkDebug(self):
        self.addBlkDebug('debug0', 'node0')
        # 'node0' is used by the blkdebug node
        self.delBlockDriverState('node0', expect_error = True)
        # But we can remove the blkdebug node directly
        self.delBlockDriverState('debug0')
        self.checkBlockDriverState('node0', False)

    def testBlkVerify(self):
        self.addBlkVerify('verify0', 'node0', 'node1')
        # We cannot remove the children of a blkverify device
        self.delBlockDriverState('node0', expect_error = True)
        self.delBlockDriverState('node1', expect_error = True)
        # But we can remove the blkverify node directly
        self.delBlockDriverState('verify0')
        self.checkBlockDriverState('node0', False)
        self.checkBlockDriverState('node1', False)

    def testQuorum(self):
        if not 'quorum' in iotests.qemu_img_pipe('--help'):
            return
        self.addQuorum('quorum0', 'node0', 'node1')
        # We cannot remove the children of a Quorum device
        self.delBlockDriverState('node0', expect_error = True)
        self.delBlockDriverState('node1', expect_error = True)
        # But we can remove the Quorum node directly
        self.delBlockDriverState('quorum0')
        self.checkBlockDriverState('node0', False)
        self.checkBlockDriverState('node1', False)


if __name__ == '__main__':
    iotests.main(supported_fmts=["qcow2"])
Commit	Line	Data
342075fd AG	1	#!/usr/bin/env python
	2	#
	3	# Test cases for the QMP 'x-blockdev-del' command
	4	#
	5	# Copyright (C) 2015 Igalia, S.L.
	6	# Author: Alberto Garcia <[email protected]>
	7	#
	8	# This program is free software; you can redistribute it and/or modify
	9	# it under the terms of the GNU General Public License as published by
	10	# the Free Software Foundation; either version 2 of the License, or
	11	# (at your option) any later version.
	12	#
	13	# This program is distributed in the hope that it will be useful,
	14	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	15	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	16	# GNU General Public License for more details.
	17	#
	18	# You should have received a copy of the GNU General Public License
	19	# along with this program. If not, see <http://www.gnu.org/licenses/>.
	20	#
	21
	22	import os
	23	import iotests
	24	import time
	25
	26	base_img = os.path.join(iotests.test_dir, 'base.img')
	27	new_img = os.path.join(iotests.test_dir, 'new.img')
	28
	29	class TestBlockdevDel(iotests.QMPTestCase):
	30
	31	def setUp(self):
	32	iotests.qemu_img('create', '-f', iotests.imgfmt, base_img, '1M')
	33	self.vm = iotests.VM()
	34	self.vm.launch()
	35
	36	def tearDown(self):
	37	self.vm.shutdown()
	38	os.remove(base_img)
	39	if os.path.isfile(new_img):
	40	os.remove(new_img)
	41
	42	# Check whether a BlockBackend exists
	43	def checkBlockBackend(self, backend, node, must_exist = True):
	44	result = self.vm.qmp('query-block')
	45	backends = filter(lambda x: x['device'] == backend, result['return'])
	46	self.assertLessEqual(len(backends), 1)
	47	self.assertEqual(must_exist, len(backends) == 1)
	48	if must_exist:
	49	if node:
	50	self.assertEqual(backends[0]['inserted']['node-name'], node)
	51	else:
	52	self.assertFalse(backends[0].has_key('inserted'))
	53
	54	# Check whether a BlockDriverState exists
	55	def checkBlockDriverState(self, node, must_exist = True):
	56	result = self.vm.qmp('query-named-block-nodes')
	57	nodes = filter(lambda x: x['node-name'] == node, result['return'])
	58	self.assertLessEqual(len(nodes), 1)
	59	self.assertEqual(must_exist, len(nodes) == 1)
	60
	61	# Add a new BlockBackend (with its attached BlockDriverState)
	62	def addBlockBackend(self, backend, node):
	63	file_node = '%s_file' % node
	64	self.checkBlockBackend(backend, node, False)
65	self.checkBlockDriverState(node, False)
66	self.checkBlockDriverState(file_node, False)
67	opts = {'driver': iotests.imgfmt,
68	'id': backend,
69	'node-name': node,
70	'file': {'driver': 'file',
71	'node-name': file_node,
72	'filename': base_img}}
73	result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
74	self.assert_qmp(result, 'return', {})
75	self.checkBlockBackend(backend, node)
76	self.checkBlockDriverState(node)
77	self.checkBlockDriverState(file_node)
78
79	# Add a BlockDriverState without a BlockBackend
80	def addBlockDriverState(self, node):
81	file_node = '%s_file' % node
82	self.checkBlockDriverState(node, False)
83	self.checkBlockDriverState(file_node, False)
84	opts = {'driver': iotests.imgfmt,
85	'node-name': node,
86	'file': {'driver': 'file',
87	'node-name': file_node,
88	'filename': base_img}}
89	result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
90	self.assert_qmp(result, 'return', {})
91	self.checkBlockDriverState(node)
92	self.checkBlockDriverState(file_node)
93
94	# Add a BlockDriverState that will be used as overlay for the base_img BDS
95	def addBlockDriverStateOverlay(self, node):
96	self.checkBlockDriverState(node, False)
97	iotests.qemu_img('create', '-f', iotests.imgfmt,
98	'-b', base_img, new_img, '1M')
99	opts = {'driver': iotests.imgfmt,
100	'node-name': node,
101	'backing': '',
102	'file': {'driver': 'file',
103	'filename': new_img}}
104	result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
105	self.assert_qmp(result, 'return', {})
106	self.checkBlockDriverState(node)
107
108	# Delete a BlockBackend
109	def delBlockBackend(self, backend, node, expect_error = False,
110	destroys_media = True):
111	self.checkBlockBackend(backend, node)
112	if node:
113	self.checkBlockDriverState(node)
114	result = self.vm.qmp('x-blockdev-del', id = backend)
115	if expect_error:
116	self.assert_qmp(result, 'error/class', 'GenericError')
117	if node:
118	self.checkBlockDriverState(node)
119	else:
120	self.assert_qmp(result, 'return', {})
121	if node:
122	self.checkBlockDriverState(node, not destroys_media)
123	self.checkBlockBackend(backend, node, must_exist = expect_error)
124
125	# Delete a BlockDriverState
126	def delBlockDriverState(self, node, expect_error = False):
127	self.checkBlockDriverState(node)
128	result = self.vm.qmp('x-blockdev-del', node_name = node)
129	if expect_error:
130	self.assert_qmp(result, 'error/class', 'GenericError')
131	else:
132	self.assert_qmp(result, 'return', {})
133	self.checkBlockDriverState(node, expect_error)
134
135	# Add a device model
136	def addDeviceModel(self, device, backend):
137	result = self.vm.qmp('device_add', id = device,
138	driver = 'virtio-blk-pci', drive = backend)
139	self.assert_qmp(result, 'return', {})
140
141	# Delete a device model
142	def delDeviceModel(self, device):
143	result = self.vm.qmp('device_del', id = device)
144	self.assert_qmp(result, 'return', {})
145
146	result = self.vm.qmp('system_reset')
147	self.assert_qmp(result, 'return', {})
148
149	device_path = '/machine/peripheral/%s/virtio-backend' % device
150	event = self.vm.event_wait(name="DEVICE_DELETED",
151	match={'data': {'path': device_path}})
152	self.assertNotEqual(event, None)
153
154	event = self.vm.event_wait(name="DEVICE_DELETED",
155	match={'data': {'device': device}})
156	self.assertNotEqual(event, None)
157
158	# Remove a BlockDriverState
159	def ejectDrive(self, backend, node, expect_error = False,
160	destroys_media = True):
161	self.checkBlockBackend(backend, node)
162	self.checkBlockDriverState(node)
163	result = self.vm.qmp('eject', device = backend)
164	if expect_error:
165	self.assert_qmp(result, 'error/class', 'GenericError')
166	self.checkBlockDriverState(node)
167	self.checkBlockBackend(backend, node)
168	else:
169	self.assert_qmp(result, 'return', {})
170	self.checkBlockDriverState(node, not destroys_media)
171	self.checkBlockBackend(backend, None)
172
173	# Insert a BlockDriverState
174	def insertDrive(self, backend, node):
175	self.checkBlockBackend(backend, None)
176	self.checkBlockDriverState(node)
6e0abc25	177	result = self.vm.qmp('x-blockdev-insert-medium',
342075fd AG	178	device = backend, node_name = node)
	179	self.assert_qmp(result, 'return', {})
	180	self.checkBlockBackend(backend, node)
	181	self.checkBlockDriverState(node)
	182
	183	# Create a snapshot using 'blockdev-snapshot-sync'
	184	def createSnapshotSync(self, node, overlay):
	185	self.checkBlockDriverState(node)
	186	self.checkBlockDriverState(overlay, False)
	187	opts = {'node-name': node,
	188	'snapshot-file': new_img,
	189	'snapshot-node-name': overlay,
	190	'format': iotests.imgfmt}
	191	result = self.vm.qmp('blockdev-snapshot-sync', conv_keys=False, **opts)
	192	self.assert_qmp(result, 'return', {})
	193	self.checkBlockDriverState(node)
	194	self.checkBlockDriverState(overlay)
	195
	196	# Create a snapshot using 'blockdev-snapshot'
	197	def createSnapshot(self, node, overlay):
	198	self.checkBlockDriverState(node)
	199	self.checkBlockDriverState(overlay)
	200	result = self.vm.qmp('blockdev-snapshot',
	201	node = node, overlay = overlay)
	202	self.assert_qmp(result, 'return', {})
	203	self.checkBlockDriverState(node)
	204	self.checkBlockDriverState(overlay)
	205
	206	# Create a mirror
	207	def createMirror(self, backend, node, new_node):
	208	self.checkBlockBackend(backend, node)
	209	self.checkBlockDriverState(new_node, False)
	210	opts = {'device': backend,
	211	'target': new_img,
	212	'node-name': new_node,
	213	'sync': 'top',
	214	'format': iotests.imgfmt}
	215	result = self.vm.qmp('drive-mirror', conv_keys=False, **opts)
	216	self.assert_qmp(result, 'return', {})
	217	self.checkBlockBackend(backend, node)
	218	self.checkBlockDriverState(new_node)
	219
	220	# Complete an existing block job
	221	def completeBlockJob(self, backend, node_before, node_after):
	222	self.checkBlockBackend(backend, node_before)
	223	result = self.vm.qmp('block-job-complete', device=backend)
	224	self.assert_qmp(result, 'return', {})
	225	self.wait_until_completed(backend)
	226	self.checkBlockBackend(backend, node_after)
	227
	228	# Add a BlkDebug node
	229	# Note that the purpose of this is to test the x-blockdev-del
	230	# sanity checks, not to create a usable blkdebug drive
	231	def addBlkDebug(self, debug, node):
	232	self.checkBlockDriverState(node, False)
	233	self.checkBlockDriverState(debug, False)
	234	image = {'driver': iotests.imgfmt,
	235	'node-name': node,
	236	'file': {'driver': 'file',
	237	'filename': base_img}}
	238	opts = {'driver': 'blkdebug',
	239	'node-name': debug,
	240	'image': image}
	241	result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
242	self.assert_qmp(result, 'return', {})
243	self.checkBlockDriverState(node)
244	self.checkBlockDriverState(debug)
245
246	# Add a BlkVerify node
247	# Note that the purpose of this is to test the x-blockdev-del
248	# sanity checks, not to create a usable blkverify drive
249	def addBlkVerify(self, blkverify, test, raw):
250	self.checkBlockDriverState(test, False)
251	self.checkBlockDriverState(raw, False)
252	self.checkBlockDriverState(blkverify, False)
253	iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M')
254	node_0 = {'driver': iotests.imgfmt,
255	'node-name': test,
256	'file': {'driver': 'file',
257	'filename': base_img}}
258	node_1 = {'driver': iotests.imgfmt,
259	'node-name': raw,
260	'file': {'driver': 'file',
261	'filename': new_img}}
262	opts = {'driver': 'blkverify',
263	'node-name': blkverify,
264	'test': node_0,
265	'raw': node_1}
266	result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
267	self.assert_qmp(result, 'return', {})
268	self.checkBlockDriverState(test)
269	self.checkBlockDriverState(raw)
270	self.checkBlockDriverState(blkverify)
271
272	# Add a Quorum node
273	def addQuorum(self, quorum, child0, child1):
274	self.checkBlockDriverState(child0, False)
275	self.checkBlockDriverState(child1, False)
276	self.checkBlockDriverState(quorum, False)
277	iotests.qemu_img('create', '-f', iotests.imgfmt, new_img, '1M')
278	child_0 = {'driver': iotests.imgfmt,
279	'node-name': child0,
280	'file': {'driver': 'file',
281	'filename': base_img}}
282	child_1 = {'driver': iotests.imgfmt,
283	'node-name': child1,
284	'file': {'driver': 'file',
285	'filename': new_img}}
286	opts = {'driver': 'quorum',
287	'node-name': quorum,
288	'vote-threshold': 1,
289	'children': [ child_0, child_1 ]}
290	result = self.vm.qmp('blockdev-add', conv_keys = False, options = opts)
291	self.assert_qmp(result, 'return', {})
292	self.checkBlockDriverState(child0)
293	self.checkBlockDriverState(child1)
294	self.checkBlockDriverState(quorum)
295
296	########################
297	# The tests start here #
298	########################
299
300	def testWrongParameters(self):
301	self.addBlockBackend('drive0', 'node0')
302	result = self.vm.qmp('x-blockdev-del')
303	self.assert_qmp(result, 'error/class', 'GenericError')
304	result = self.vm.qmp('x-blockdev-del', id='drive0', node_name='node0')
305	self.assert_qmp(result, 'error/class', 'GenericError')
306	self.delBlockBackend('drive0', 'node0')
307
308	def testBlockBackend(self):
309	self.addBlockBackend('drive0', 'node0')
310	# You cannot delete a BDS that is attached to a backend
311	self.delBlockDriverState('node0', expect_error = True)
312	self.delBlockBackend('drive0', 'node0')
313
314	def testBlockDriverState(self):
315	self.addBlockDriverState('node0')
316	# You cannot delete a file BDS directly
317	self.delBlockDriverState('node0_file', expect_error = True)
318	self.delBlockDriverState('node0')
319
320	def testEject(self):
321	self.addBlockBackend('drive0', 'node0')
322	self.ejectDrive('drive0', 'node0')
323	self.delBlockBackend('drive0', None)
324
325	def testDeviceModel(self):
326	self.addBlockBackend('drive0', 'node0')
327	self.addDeviceModel('device0', 'drive0')
328	self.ejectDrive('drive0', 'node0', expect_error = True)
329	self.delBlockBackend('drive0', 'node0', expect_error = True)
330	self.delDeviceModel('device0')
331	self.delBlockBackend('drive0', 'node0')
332
333	def testAttachMedia(self):
334	# This creates a BlockBackend and removes its media
335	self.addBlockBackend('drive0', 'node0')
336	self.ejectDrive('drive0', 'node0')
337	# This creates a new BlockDriverState and inserts it into the backend
338	self.addBlockDriverState('node1')
339	self.insertDrive('drive0', 'node1')
340	# The backend can't be removed: the new BDS has an extra reference
341	self.delBlockBackend('drive0', 'node1', expect_error = True)
342	self.delBlockDriverState('node1', expect_error = True)
343	# The BDS still exists after being ejected, but now it can be removed
344	self.ejectDrive('drive0', 'node1', destroys_media = False)
345	self.delBlockDriverState('node1')
346	self.delBlockBackend('drive0', None)
347
348	def testSnapshotSync(self):
349	self.addBlockBackend('drive0', 'node0')
350	self.createSnapshotSync('node0', 'overlay0')
351	# This fails because node0 is now being used as a backing image
352	self.delBlockDriverState('node0', expect_error = True)
353	# This succeeds because overlay0 only has the backend reference
354	self.delBlockBackend('drive0', 'overlay0')
355	self.checkBlockDriverState('node0', False)
356
357	def testSnapshot(self):
358	self.addBlockBackend('drive0', 'node0')
359	self.addBlockDriverStateOverlay('overlay0')
360	self.createSnapshot('node0', 'overlay0')
361	self.delBlockBackend('drive0', 'overlay0', expect_error = True)
362	self.delBlockDriverState('node0', expect_error = True)
363	self.delBlockDriverState('overlay0', expect_error = True)
364	self.ejectDrive('drive0', 'overlay0', destroys_media = False)
365	self.delBlockBackend('drive0', None)
366	self.delBlockDriverState('node0', expect_error = True)
367	self.delBlockDriverState('overlay0')
368	self.checkBlockDriverState('node0', False)
369
370	def testMirror(self):
371	self.addBlockBackend('drive0', 'node0')
372	self.createMirror('drive0', 'node0', 'mirror0')
373	# The block job prevents removing the device
374	self.delBlockBackend('drive0', 'node0', expect_error = True)
375	self.delBlockDriverState('node0', expect_error = True)
376	self.delBlockDriverState('mirror0', expect_error = True)
377	self.wait_ready('drive0')
378	self.completeBlockJob('drive0', 'node0', 'mirror0')
379	self.assert_no_active_block_jobs()
380	self.checkBlockDriverState('node0', False)
381	# This succeeds because the backend now points to mirror0
382	self.delBlockBackend('drive0', 'mirror0')
383
384	def testBlkDebug(self):
385	self.addBlkDebug('debug0', 'node0')
386	# 'node0' is used by the blkdebug node
387	self.delBlockDriverState('node0', expect_error = True)
388	# But we can remove the blkdebug node directly
389	self.delBlockDriverState('debug0')
390	self.checkBlockDriverState('node0', False)
391
392	def testBlkVerify(self):
393	self.addBlkVerify('verify0', 'node0', 'node1')
394	# We cannot remove the children of a blkverify device
395	self.delBlockDriverState('node0', expect_error = True)
396	self.delBlockDriverState('node1', expect_error = True)
397	# But we can remove the blkverify node directly
398	self.delBlockDriverState('verify0')
399	self.checkBlockDriverState('node0', False)
400	self.checkBlockDriverState('node1', False)
401
402	def testQuorum(self):
92e68987 AG	403	if not 'quorum' in iotests.qemu_img_pipe('--help'):
92e68987 AG	404	return
342075fd AG	405	self.addQuorum('quorum0', 'node0', 'node1')
	406	# We cannot remove the children of a Quorum device
	407	self.delBlockDriverState('node0', expect_error = True)
	408	self.delBlockDriverState('node1', expect_error = True)
	409	# But we can remove the Quorum node directly
	410	self.delBlockDriverState('quorum0')
	411	self.checkBlockDriverState('node0', False)
	412	self.checkBlockDriverState('node1', False)
	413
	414
	415	if __name__ == '__main__':
	416	iotests.main(supported_fmts=["qcow2"])