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         result = self.vm.qmp('nbd-server-start', addr=address)
  48         self.assert_qmp(result, 'return', {})
  49         result = self.vm.qmp('nbd-server-add', device='drive0', name='exp')
  50         self.assert_qmp(result, 'return', {})
  51
  52     def tearDown(self):
  53         self.vm.shutdown()
  54         os.remove(nbd_sock)
  55         os.remove(disk)
  56
  57     def remove_export(self, name, mode=None):
  58         if mode is None:
  59             return self.vm.qmp('nbd-server-remove', name=name)
  60         else:
  61             return self.vm.qmp('nbd-server-remove', name=name, mode=mode)
  62
  63     def assertExportNotFound(self, name):
  64         result = self.vm.qmp('nbd-server-remove', name=name)
  65         self.assert_qmp(result, 'error/desc', "Export 'exp' is not found")
  66
  67     def assertExistingClients(self, result):
  68         self.assert_qmp(result, 'error/desc', "export 'exp' still in use")
  69
  70     def assertReadOk(self, qemu_io_output):
  71         self.assertEqual(
  72                 filter_qemu_io(qemu_io_output).strip(),
  73                 'read 512/512 bytes at offset 0\n' +
  74                 '512 bytes, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)')
  75
  76     def assertReadFailed(self, qemu_io_output):
  77         self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
  78                          'read failed: Input/output error')
  79
  80     def assertConnectFailed(self, qemu_io_output):
  81         self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
  82                          "qemu-io: can't open device " + nbd_uri +
  83                          ": Requested export not available\n"
  84                          "server reported: export 'exp' not present")
  85
  86     def do_test_connect_after_remove(self, mode=None):
  87         args = ('-r', '-f', 'raw', '-c', 'read 0 512', nbd_uri)
  88         self.assertReadOk(qemu_io(*args))
  89
  90         result = self.remove_export('exp', mode)
  91         self.assert_qmp(result, 'return', {})
  92
  93         self.assertExportNotFound('exp')
  94         self.assertConnectFailed(qemu_io(*args))
  95
  96     def test_connect_after_remove_default(self):
  97         self.do_test_connect_after_remove()
  98
  99     def test_connect_after_remove_safe(self):
 100         self.do_test_connect_after_remove('safe')
 101
 102     def test_connect_after_remove_force(self):
 103         self.do_test_connect_after_remove('hard')
 104
 105     def do_test_remove_during_connect_safe(self, mode=None):
 106         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
 107         self.assertReadOk(qio.cmd('read 0 512'))
 108
 109         result = self.remove_export('exp', mode)
 110         self.assertExistingClients(result)
 111
 112         self.assertReadOk(qio.cmd('read 0 512'))
 113
 114         qio.close()
 115
 116         result = self.remove_export('exp', mode)
 117         self.assert_qmp(result, 'return', {})
 118
 119         self.assertExportNotFound('exp')
 120
 121     def test_remove_during_connect_default(self):
 122         self.do_test_remove_during_connect_safe()
 123
 124     def test_remove_during_connect_safe(self):
 125         self.do_test_remove_during_connect_safe('safe')
 126
 127     def test_remove_during_connect_hard(self):
 128         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
 129         self.assertReadOk(qio.cmd('read 0 512'))
 130
 131         result = self.remove_export('exp', 'hard')
 132         self.assert_qmp(result, 'return', {})
 133
 134         self.assertReadFailed(qio.cmd('read 0 512'))
 135         self.assertExportNotFound('exp')
 136
 137         qio.close()
 138
 139     def test_remove_during_connect_safe_hard(self):
 140         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
 141         self.assertReadOk(qio.cmd('read 0 512'))
 142
 143         result = self.remove_export('exp', 'safe')
 144         self.assertExistingClients(result)
 145
 146         self.assertReadOk(qio.cmd('read 0 512'))
 147
 148         result = self.remove_export('exp', 'hard')
 149         self.assert_qmp(result, 'return', {})
 150
 151         self.assertExportNotFound('exp')
 152         self.assertReadFailed(qio.cmd('read 0 512'))
 153         qio.close()
 154
 155
 156 if __name__ == '__main__':
 157     iotests.main(supported_fmts=['raw'],
 158                  supported_protocols=['nbd'])