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']) 157