205 (5340B)
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).stdout) 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, check=False).stdout) 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'])