path: root/tests/qemu-iotests/118

#!/usr/bin/env python3
# group: rw
#
# Test case for media change monitor commands
#
# Copyright (C) 2015 Red Hat, Inc.
#
# 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 stat
import time
import iotests
from iotests import qemu_img

old_img = os.path.join(iotests.test_dir, 'test0.img')
new_img = os.path.join(iotests.test_dir, 'test1.img')

def interface_to_device_name(interface):
    if interface == 'ide':
        return 'ide-cd'
    elif interface == 'floppy':
        return 'floppy'
    elif interface == 'scsi':
        return 'scsi-cd'
    else:
        return None

class ChangeBaseClass(iotests.QMPTestCase):
    has_opened = False
    has_closed = False

    device_name = 'qdev0'
    use_drive = False

    def process_events(self):
        for event in self.vm.get_qmp_events(wait=False):
            if (event['event'] == 'DEVICE_TRAY_MOVED' and
                (event['data']['device'] == 'drive0' or
                 event['data']['id'] == self.device_name)):
                if event['data']['tray-open'] == False:
                    self.has_closed = True
                else:
                    self.has_opened = True

    def wait_for_open(self):
        if not self.has_real_tray:
            return

        with iotests.Timeout(3, 'Timeout while waiting for the tray to open'):
            while not self.has_opened:
                self.process_events()

    def wait_for_close(self):
        if not self.has_real_tray:
            return

        with iotests.Timeout(3, 'Timeout while waiting for the tray to close'):
            while not self.has_closed:
                self.process_events()

class GeneralChangeTestsBaseClass(ChangeBaseClass):

    def test_blockdev_change_medium(self):
        self.vm.cmd('blockdev-change-medium',
                    id=self.device_name, filename=new_img,
                    format=iotests.imgfmt)

        self.wait_for_open()
        self.wait_for_close()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    def test_eject(self):
        self.vm.cmd('eject', id=self.device_name, force=True)

        self.wait_for_open()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        self.assert_qmp_absent(result, 'return[0]/inserted')

    def test_tray_eject_change(self):
        self.vm.cmd('eject', id=self.device_name, force=True)

        self.wait_for_open()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        self.assert_qmp_absent(result, 'return[0]/inserted')

        self.vm.cmd('blockdev-change-medium', id=self.device_name,
                    filename=new_img, format=iotests.imgfmt)

        self.wait_for_close()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    def test_tray_open_close(self):
        self.vm.cmd('blockdev-open-tray',
                    id=self.device_name, force=True)

        self.wait_for_open()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        if self.was_empty == True:
            self.assert_qmp_absent(result, 'return[0]/inserted')
        else:
            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-close-tray', id=self.device_name)

        if self.has_real_tray or not self.was_empty:
            self.wait_for_close()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', False)
        if self.was_empty == True:
            self.assert_qmp_absent(result, 'return[0]/inserted')
        else:
            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

    def test_tray_eject_close(self):
        self.vm.cmd('eject', id=self.device_name, force=True)

        self.wait_for_open()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        self.assert_qmp_absent(result, 'return[0]/inserted')

        self.vm.cmd('blockdev-close-tray', id=self.device_name)

        self.wait_for_close()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', False)
        self.assert_qmp_absent(result, 'return[0]/inserted')

    def test_tray_open_change(self):
        self.vm.cmd('blockdev-open-tray', id=self.device_name,
                                          force=True)

        self.wait_for_open()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        if self.was_empty == True:
            self.assert_qmp_absent(result, 'return[0]/inserted')
        else:
            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-change-medium', id=self.device_name,
                                              filename=new_img,
                                              format=iotests.imgfmt)

        self.wait_for_close()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    def test_cycle(self, read_only_node=False):
        self.vm.cmd('blockdev-add',
                    node_name='new',
                    driver=iotests.imgfmt,
                    read_only=read_only_node,
                    file={'filename': new_img,
                           'driver': 'file'})

        self.vm.cmd('blockdev-open-tray',
                    id=self.device_name, force=True)

        self.wait_for_open()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        if self.was_empty == True:
            self.assert_qmp_absent(result, 'return[0]/inserted')
        else:
            self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-remove-medium',
                    id=self.device_name)

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        self.assert_qmp_absent(result, 'return[0]/inserted')

        self.vm.cmd('blockdev-insert-medium',
                    id=self.device_name, node_name='new')

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

        self.vm.cmd('blockdev-close-tray', id=self.device_name)

        self.wait_for_close()

        result = self.vm.qmp('query-block')
        if self.has_real_tray:
            self.assert_qmp(result, 'return[0]/tray_open', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    def test_cycle_read_only_media(self):
        self.test_cycle(True)

    def test_close_on_closed(self):
        # Should be a no-op
        self.vm.cmd('blockdev-close-tray', id=self.device_name)
        self.assertEqual(self.vm.get_qmp_events(wait=False), [])

    def test_remove_on_closed(self):
        if not self.has_real_tray:
            return

        result = self.vm.qmp('blockdev-remove-medium', id=self.device_name)
        self.assert_qmp(result, 'error/class', 'GenericError')

    def test_insert_on_closed(self):
        if not self.has_real_tray:
            return

        self.vm.cmd('blockdev-add',
                    node_name='new',
                    driver=iotests.imgfmt,
                    file={'filename': new_img,
                          'driver': 'file'})

        result = self.vm.qmp('blockdev-insert-medium', id=self.device_name,
                                                       node_name='new')
        self.assert_qmp(result, 'error/class', 'GenericError')

class TestInitiallyFilled(GeneralChangeTestsBaseClass):
    was_empty = False

    def setUp(self):
        qemu_img('create', '-f', iotests.imgfmt, old_img, '1440k')
        qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')
        self.vm = iotests.VM()
        if self.use_drive:
            self.vm.add_drive(old_img, 'media=%s' % self.media, 'none')
        else:
            self.vm.add_blockdev([ 'node-name=drive0',
                                   'driver=%s' % iotests.imgfmt,
                                   'file.driver=file',
                                   'file.filename=%s' % old_img ])
        if self.interface == 'scsi':
            self.vm.add_object('iothread,id=iothread0')
            self.vm.add_device('virtio-scsi-pci,iothread=iothread0')
        self.vm.add_device('%s,drive=drive0,id=%s' %
                           (interface_to_device_name(self.interface),
                            self.device_name))
        self.vm.launch()

    def tearDown(self):
        self.vm.shutdown()
        os.remove(old_img)
        os.remove(new_img)

    def test_insert_on_filled(self):
        self.vm.cmd('blockdev-add',
                    node_name='new',
                    driver=iotests.imgfmt,
                    file={'filename': new_img,
                          'driver': 'file'})

        self.vm.cmd('blockdev-open-tray', id=self.device_name)

        self.wait_for_open()

        result = self.vm.qmp('blockdev-insert-medium', id=self.device_name,
                                                       node_name='new')
        self.assert_qmp(result, 'error/class', 'GenericError')

class TestInitiallyEmpty(GeneralChangeTestsBaseClass):
    was_empty = True

    def setUp(self):
        qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')
        self.vm = iotests.VM()
        if self.use_drive:
            self.vm.add_drive(None, 'media=%s' % self.media, 'none')
        if self.interface == 'scsi':
            self.vm.add_object('iothread,id=iothread0')
            self.vm.add_device('virtio-scsi-pci,iothread=iothread0')
        self.vm.add_device('%s,%sid=%s' %
                           (interface_to_device_name(self.interface),
                            'drive=drive0,' if self.use_drive else '',
                            self.device_name))
        self.vm.launch()

    def tearDown(self):
        self.vm.shutdown()
        os.remove(new_img)

    def test_remove_on_empty(self):
        self.vm.cmd('blockdev-open-tray', id=self.device_name)

        self.wait_for_open()

        # Should be a no-op
        self.vm.cmd('blockdev-remove-medium', id=self.device_name)

# Do this in a function to avoid leaking variables like case into the global
# name space (otherwise tests would be run for the abstract base classes)
def create_basic_test_classes():
    for (media, interface, has_real_tray) in [ ('cdrom', 'ide', True),
                                               ('cdrom', 'scsi', True),
                                               ('disk', 'floppy', False) ]:

        for case in [ TestInitiallyFilled, TestInitiallyEmpty ]:
            for use_drive in [ True, False ]:
                attr = { 'media': media,
                         'interface': interface,
                         'has_real_tray': has_real_tray,
                         'use_drive': use_drive }

                name = '%s_%s_%s_%s' % (case.__name__, media, interface,
                                        'drive' if use_drive else 'blockdev')
                globals()[name] = type(name, (case, ), attr)

create_basic_test_classes()

class TestChangeReadOnly(ChangeBaseClass):
    device_name = 'qdev0'

    def setUp(self):
        qemu_img('create', '-f', iotests.imgfmt, old_img, '1440k')
        qemu_img('create', '-f', iotests.imgfmt, new_img, '1440k')
        self.vm = iotests.VM()

    def tearDown(self):
        self.vm.shutdown()
        os.chmod(old_img, 0o666)
        os.chmod(new_img, 0o666)
        os.remove(old_img)
        os.remove(new_img)

    def test_ro_ro_retain(self):
        os.chmod(old_img, 0o444)
        os.chmod(new_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-change-medium', id=self.device_name,
                                              filename=new_img,
                                              format=iotests.imgfmt,
                                              read_only_mode='retain')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    def test_ro_rw_retain(self):
        os.chmod(old_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-change-medium', id=self.device_name,
                                              filename=new_img,
                                              format=iotests.imgfmt,
                                              read_only_mode='retain')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    @iotests.skip_if_user_is_root
    def test_rw_ro_retain(self):
        os.chmod(new_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        result = self.vm.qmp('blockdev-change-medium', id=self.device_name,
                                                       filename=new_img,
                                                       format=iotests.imgfmt,
                                                       read_only_mode='retain')
        self.assert_qmp(result, 'error/class', 'GenericError')

        self.assertEqual(self.vm.get_qmp_events(wait=False), [])

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

    def test_ro_rw(self):
        os.chmod(old_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-change-medium',
                    id=self.device_name,
                    filename=new_img,
                    format=iotests.imgfmt,
                    read_only_mode='read-write')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    def test_rw_ro(self):
        os.chmod(new_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-change-medium',
                    id=self.device_name,
                    filename=new_img,
                    format=iotests.imgfmt,
                    read_only_mode='read-only')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    def test_make_rw_ro(self):
        self.vm.add_drive(old_img, 'media=disk', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-change-medium',
                    id=self.device_name,
                    filename=new_img,
                    format=iotests.imgfmt,
                    read_only_mode='read-only')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    @iotests.skip_if_user_is_root
    def test_make_ro_rw(self):
        os.chmod(new_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        result = self.vm.qmp('blockdev-change-medium',
                             id=self.device_name,
                             filename=new_img,
                             format=iotests.imgfmt,
                             read_only_mode='read-write')
        self.assert_qmp(result, 'error/class', 'GenericError')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

    def test_make_rw_ro_by_retain(self):
        os.chmod(old_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk,read-only=on', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-change-medium', id=self.device_name,
                                              filename=new_img,
                                              format=iotests.imgfmt,
                                              read_only_mode='retain')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

    @iotests.skip_if_user_is_root
    def test_make_ro_rw_by_retain(self):
        os.chmod(new_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        result = self.vm.qmp('blockdev-change-medium', id=self.device_name,
                                                       filename=new_img,
                                                       format=iotests.imgfmt,
                                                       read_only_mode='retain')
        self.assert_qmp(result, 'error/class', 'GenericError')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

    def test_rw_ro_cycle(self):
        os.chmod(new_img, 0o444)
        self.vm.add_drive(old_img, 'media=disk', 'none')
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-add',
                    node_name='new',
                    driver=iotests.imgfmt,
                    read_only=True,
                    file={'filename': new_img,
                           'driver': 'file'})

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', False)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

        self.vm.cmd('blockdev-remove-medium', id=self.device_name)

        result = self.vm.qmp('query-block')
        self.assert_qmp_absent(result, 'return[0]/inserted')

        self.vm.cmd('blockdev-insert-medium', id=self.device_name,
                                              node_name='new')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/ro', True)
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)

GeneralChangeTestsBaseClass = None
TestInitiallyFilled = None
TestInitiallyEmpty = None


class TestBlockJobsAfterCycle(ChangeBaseClass):
    device_name = 'qdev0'

    def setUp(self):
        qemu_img('create', '-f', iotests.imgfmt, old_img, '1440K')

        self.vm = iotests.VM()
        self.vm.add_drive_raw("id=drive0,driver=null-co,if=none")
        self.vm.add_device('floppy,drive=drive0,id=%s' % self.device_name)
        self.vm.launch()

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/image/format', 'null-co')

        # For device-less BBs, calling blockdev-open-tray or blockdev-close-tray
        # is not necessary
        self.vm.cmd('blockdev-remove-medium', id=self.device_name)

        result = self.vm.qmp('query-block')
        self.assert_qmp_absent(result, 'return[0]/inserted')

        self.vm.cmd('blockdev-add',
                    node_name='node0',
                    driver=iotests.imgfmt,
                    file={'filename': old_img,
                          'driver': 'file'})

        self.vm.cmd('blockdev-insert-medium', id=self.device_name,
                                              node_name='node0')

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/image/filename', old_img)

    def tearDown(self):
        self.vm.shutdown()
        os.remove(old_img)
        try:
            os.remove(new_img)
        except OSError:
            pass

    # We need backing file support
    @iotests.skip_for_formats(('vpc', 'parallels', 'qcow', 'vdi', 'vmdk', 'raw',
                               'vhdx'))
    def test_snapshot_and_commit(self):
        self.vm.cmd('blockdev-snapshot-sync', device='drive0',
                                              snapshot_file=new_img,
                                              format=iotests.imgfmt)

        result = self.vm.qmp('query-block')
        self.assert_qmp(result, 'return[0]/inserted/image/filename', new_img)
        self.assert_qmp(result,
                        'return[0]/inserted/image/backing-image/filename',
                        old_img)

        self.vm.cmd('block-commit', device='drive0')

        self.vm.event_wait(name='BLOCK_JOB_READY')

        result = self.vm.qmp('query-block-jobs')
        self.assert_qmp(result, 'return[0]/device', 'drive0')

        self.vm.cmd('block-job-complete', device='drive0')

        self.vm.event_wait(name='BLOCK_JOB_COMPLETED')


if __name__ == '__main__':
    if iotests.qemu_default_machine != 'pc':
        # We need floppy and IDE CD-ROM
        iotests.notrun('not suitable for this machine type: %s' %
                       iotests.qemu_default_machine)
    # Need to support image creation
    iotests.main(supported_fmts=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2',
                                 'vmdk', 'raw', 'vhdx', 'qed'],
                 supported_protocols=['file'])