tests/qemu-iotests/205

   1 #!/usr/bin/env python3
   2 # group: rw quick
   3 #
   4 # Tests for qmp command nbd-server-remove.
   5 #
   6 # Copyright (c) 2017 Virtuozzo International GmbH
   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 sys
  24 import iotests
  25 import time
  26 from iotests import qemu_img_create, qemu_io, filter_qemu_io, QemuIoInteractive
  27
  28 nbd_sock = os.path.join(iotests.sock_dir, 'nbd_sock')
  29 nbd_uri = 'nbd+unix:///exp?socket=' + nbd_sock
  30 disk = os.path.join(iotests.test_dir, 'disk')
  31
  32
  33 class TestNbdServerRemove(iotests.QMPTestCase):
  34     def setUp(self):
  35         qemu_img_create('-f', iotests.imgfmt, disk, '1M')
  36
  37         self.vm = iotests.VM().add_drive(disk)
  38         self.vm.launch()
  39
  40         address = {
  41             'type': 'unix',
  42             'data': {
  43                 'path': nbd_sock
  44             }
  45         }
  46
  47         self.vm.cmd('nbd-server-start', addr=address)
  48         self.vm.cmd('nbd-server-add', device='drive0', name='exp')
  49
  50     def tearDown(self):
  51         self.vm.shutdown()
  52         os.remove(nbd_sock)
  53         os.remove(disk)
  54
  55     def remove_export(self, name, mode=None):
  56         if mode is None:
  57             return self.vm.qmp('nbd-server-remove', name=name)
  58         else:
  59             return self.vm.qmp('nbd-server-remove', name=name, mode=mode)
  60
  61     def assertExportNotFound(self, name):
  62         result = self.vm.qmp('nbd-server-remove', name=name)
  63         self.assert_qmp(result, 'error/desc', "Export 'exp' is not found")
  64
  65     def assertExistingClients(self, result):
  66         self.assert_qmp(result, 'error/desc', "export 'exp' still in use")
  67
  68     def assertReadOk(self, qemu_io_output):
  69         self.assertEqual(
  70                 filter_qemu_io(qemu_io_output).strip(),
  71                 'read 512/512 bytes at offset 0\n' +
  72                 '512 bytes, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)')
  73
  74     def assertReadFailed(self, qemu_io_output):
  75         self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
  76                          'read failed: Input/output error')
  77
  78     def assertConnectFailed(self, qemu_io_output):
  79         self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
  80                          "qemu-io: can't open device " + nbd_uri +
  81                          ": Requested export not available\n"
  82                          "server reported: export 'exp' not present")
  83
  84     def do_test_connect_after_remove(self, mode=None):
  85         args = ('-r', '-f', 'raw', '-c', 'read 0 512', nbd_uri)
  86         self.assertReadOk(qemu_io(*args).stdout)
  87
  88         result = self.remove_export('exp', mode)
  89         self.assert_qmp(result, 'return', {})
  90
  91         self.assertExportNotFound('exp')
  92         self.assertConnectFailed(qemu_io(*args, check=False).stdout)
  93
  94     def test_connect_after_remove_default(self):
  95         self.do_test_connect_after_remove()
  96
  97     def test_connect_after_remove_safe(self):
  98         self.do_test_connect_after_remove('safe')
  99
 100     def test_connect_after_remove_force(self):
 101         self.do_test_connect_after_remove('hard')
 102
 103     def do_test_remove_during_connect_safe(self, mode=None):
 104         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
 105         self.assertReadOk(qio.cmd('read 0 512'))
 106
 107         result = self.remove_export('exp', mode)
 108         self.assertExistingClients(result)
 109
 110         self.assertReadOk(qio.cmd('read 0 512'))
 111
 112         qio.close()
 113
 114         result = self.remove_export('exp', mode)
 115         self.assert_qmp(result, 'return', {})
 116
 117         self.assertExportNotFound('exp')
 118
 119     def test_remove_during_connect_default(self):
 120         self.do_test_remove_during_connect_safe()
 121
 122     def test_remove_during_connect_safe(self):
 123         self.do_test_remove_during_connect_safe('safe')
 124
 125     def test_remove_during_connect_hard(self):
 126         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
 127         self.assertReadOk(qio.cmd('read 0 512'))
 128
 129         result = self.remove_export('exp', 'hard')
 130         self.assert_qmp(result, 'return', {})
 131
 132         self.assertReadFailed(qio.cmd('read 0 512'))
 133         self.assertExportNotFound('exp')
 134
 135         qio.close()
 136
 137     def test_remove_during_connect_safe_hard(self):
 138         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
 139         self.assertReadOk(qio.cmd('read 0 512'))
 140
 141         result = self.remove_export('exp', 'safe')
 142         self.assertExistingClients(result)
 143
 144         self.assertReadOk(qio.cmd('read 0 512'))
 145
 146         result = self.remove_export('exp', 'hard')
 147         self.assert_qmp(result, 'return', {})
 148
 149         self.assertExportNotFound('exp')
 150         self.assertReadFailed(qio.cmd('read 0 512'))
 151         qio.close()
 152
 153
 154 if __name__ == '__main__':
 155     iotests.main(supported_fmts=['raw'],
 156                  supported_protocols=['nbd'])