xemu/tests/qemu-iotests/240

#!/usr/bin/env python3
# group: quick

# Test hot plugging and unplugging with iothreads
#
# Copyright (C) 2019 Igalia, S.L.
# Author: Alberto Garcia <berto@igalia.com>
#
# 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 iotests
import os

nbd_sock = iotests.file_path('nbd.sock', base_dir=iotests.sock_dir)

class TestCase(iotests.QMPTestCase):
    test_driver = "null-co"

    def required_drivers(self):
        return [self.test_driver]

    @iotests.skip_if_unsupported(required_drivers)
    def setUp(self):
        self.vm = iotests.VM()
        self.vm.launch()

    def tearDown(self):
        self.vm.shutdown()

    def test1(self):
        iotests.log('==Unplug a SCSI disk and then plug it again==')
        self.vm.qmp_log('blockdev-add', driver='null-co', read_zeroes=True, node_name='hd0')
        self.vm.qmp_log('object-add', qom_type='iothread', id="iothread0")
        self.vm.qmp_log('device_add', id='scsi0', driver='virtio-scsi', iothread='iothread0', filters=[iotests.filter_qmp_virtio_scsi])
        self.vm.qmp_log('device_add', id='scsi-hd0', driver='scsi-hd', drive='hd0')
        self.vm.qmp_log('device_del', id='scsi-hd0')
        self.vm.event_wait('DEVICE_DELETED')
        self.vm.qmp_log('device_add', id='scsi-hd0', driver='scsi-hd', drive='hd0')
        self.vm.qmp_log('device_del', id='scsi-hd0')
        self.vm.event_wait('DEVICE_DELETED')
        self.vm.qmp_log('blockdev-del', node_name='hd0')

    def test2(self):
        iotests.log('==Attach two SCSI disks using the same block device and the same iothread==')
        self.vm.qmp_log('blockdev-add', driver='null-co', read_zeroes=True, node_name='hd0', read_only=True)
        self.vm.qmp_log('object-add', qom_type='iothread', id="iothread0")
        self.vm.qmp_log('device_add', id='scsi0', driver='virtio-scsi', iothread='iothread0', filters=[iotests.filter_qmp_virtio_scsi])

        self.vm.qmp_log('device_add', id='scsi-hd0', driver='scsi-hd', drive='hd0')
        self.vm.qmp_log('device_add', id='scsi-hd1', driver='scsi-hd', drive='hd0')
        self.vm.qmp_log('device_del', id='scsi-hd0')
        self.vm.event_wait('DEVICE_DELETED')
        self.vm.qmp_log('device_del', id='scsi-hd1')
        self.vm.event_wait('DEVICE_DELETED')
        self.vm.qmp_log('blockdev-del', node_name='hd0')

    def test3(self):
        iotests.log('==Attach two SCSI disks using the same block device but different iothreads==')

        self.vm.qmp_log('blockdev-add', driver='null-co', read_zeroes=True, node_name='hd0', read_only=True)

        self.vm.qmp_log('object-add', qom_type='iothread', id="iothread0")
        self.vm.qmp_log('object-add', qom_type='iothread', id="iothread1")

        self.vm.qmp_log('device_add', id='scsi0', driver='virtio-scsi', iothread='iothread0', filters=[iotests.filter_qmp_virtio_scsi])
        self.vm.qmp_log('device_add', id='scsi1', driver='virtio-scsi', iothread='iothread1', filters=[iotests.filter_qmp_virtio_scsi])

        self.vm.qmp_log('device_add', id='scsi-hd0', driver='scsi-hd', drive='hd0', bus="scsi0.0")
        self.vm.qmp_log('device_add', id='scsi-hd1', driver='scsi-hd', drive='hd0', bus="scsi1.0")

        self.vm.qmp_log('device_del', id='scsi-hd0')
        self.vm.event_wait('DEVICE_DELETED')
        self.vm.qmp_log('device_add', id='scsi-hd1', driver='scsi-hd', drive='hd0', bus="scsi1.0")

        self.vm.qmp_log('device_del', id='scsi-hd1')
        self.vm.event_wait('DEVICE_DELETED')
        self.vm.qmp_log('blockdev-del', node_name='hd0')

    def test4(self):
        iotests.log('==Attach a SCSI disks using the same block device as a NBD server==')

        self.vm.qmp_log('blockdev-add', driver='null-co', read_zeroes=True, node_name='hd0', read_only=True)

        self.vm.qmp_log('nbd-server-start',
                        filters=[iotests.filter_qmp_testfiles],
                        addr={'type':'unix', 'data':{'path':nbd_sock}})

        self.vm.qmp_log('nbd-server-add', device='hd0')

        self.vm.qmp_log('object-add', qom_type='iothread', id="iothread0")
        self.vm.qmp_log('device_add', id='scsi0', driver='virtio-scsi', iothread='iothread0', filters=[iotests.filter_qmp_virtio_scsi])
        self.vm.qmp_log('device_add', id='scsi-hd0', driver='scsi-hd', drive='hd0')

if __name__ == '__main__':
    iotests.activate_logging()
    iotests.main()