1 #!/usr/bin/env python3 2 # group: rw auto quick 3 # 4 # Test cases for NBD multi-conn advertisement 5 # 6 # Copyright (C) 2022 Red Hat, Inc. 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 import os 22 from contextlib import contextmanager 23 from types import ModuleType 24 25 import iotests 26 from iotests import qemu_img_create, qemu_io 27 28 29 disk = os.path.join(iotests.test_dir, 'disk') 30 size = '4M' 31 nbd_sock = os.path.join(iotests.sock_dir, 'nbd_sock') 32 nbd_uri = 'nbd+unix:///{}?socket=' + nbd_sock 33 nbd: ModuleType 34 35 @contextmanager 36 def open_nbd(export_name): 37 h = nbd.NBD() 38 try: 39 h.connect_uri(nbd_uri.format(export_name)) 40 yield h 41 finally: 42 h.shutdown() 43 44 class TestNbdMulticonn(iotests.QMPTestCase): 45 def setUp(self): 46 qemu_img_create('-f', iotests.imgfmt, disk, size) 47 qemu_io('-c', 'w -P 1 0 2M', '-c', 'w -P 2 2M 2M', disk) 48 49 self.vm = iotests.VM() 50 self.vm.launch() 51 self.vm.cmd('blockdev-add', { 52 'driver': 'qcow2', 53 'node-name': 'n', 54 'file': {'driver': 'file', 'filename': disk} 55 }) 56 57 def tearDown(self): 58 self.vm.shutdown() 59 os.remove(disk) 60 try: 61 os.remove(nbd_sock) 62 except OSError: 63 pass 64 65 @contextmanager 66 def run_server(self, max_connections=None): 67 args = { 68 'addr': { 69 'type': 'unix', 70 'data': {'path': nbd_sock} 71 } 72 } 73 if max_connections is not None: 74 args['max-connections'] = max_connections 75 76 self.vm.cmd('nbd-server-start', args) 77 yield 78 79 self.vm.cmd('nbd-server-stop') 80 81 def add_export(self, name, writable=None): 82 args = { 83 'type': 'nbd', 84 'id': name, 85 'node-name': 'n', 86 'name': name, 87 } 88 if writable is not None: 89 args['writable'] = writable 90 91 self.vm.cmd('block-export-add', args) 92 93 def test_default_settings(self): 94 with self.run_server(): 95 self.add_export('r') 96 self.add_export('w', writable=True) 97 with open_nbd('r') as h: 98 self.assertTrue(h.can_multi_conn()) 99 with open_nbd('w') as h: 100 self.assertTrue(h.can_multi_conn()) 101 102 def test_limited_connections(self): 103 with self.run_server(max_connections=1): 104 self.add_export('r') 105 self.add_export('w', writable=True) 106 with open_nbd('r') as h: 107 self.assertFalse(h.can_multi_conn()) 108 with open_nbd('w') as h: 109 self.assertFalse(h.can_multi_conn()) 110 111 def test_parallel_writes(self): 112 with self.run_server(): 113 self.add_export('w', writable=True) 114 115 clients = [nbd.NBD() for _ in range(3)] 116 for c in clients: 117 c.connect_uri(nbd_uri.format('w')) 118 self.assertTrue(c.can_multi_conn()) 119 120 initial_data = clients[0].pread(1024 * 1024, 0) 121 self.assertEqual(initial_data, b'\x01' * 1024 * 1024) 122 123 updated_data = b'\x03' * 1024 * 1024 124 clients[1].pwrite(updated_data, 0) 125 clients[2].flush() 126 current_data = clients[0].pread(1024 * 1024, 0) 127 128 self.assertEqual(updated_data, current_data) 129 130 for i in range(3): 131 clients[i].shutdown() 132 133 134 if __name__ == '__main__': 135 try: 136 # Easier to use libnbd than to try and set up parallel 137 # 'qemu-nbd --list' or 'qemu-io' processes, but not all systems 138 # have libnbd installed. 139 import nbd # type: ignore 140 141 iotests.main(supported_fmts=['qcow2']) 142 except ImportError: 143 iotests.notrun('Python bindings to libnbd are not installed') 144