4 # Test cases for NBD multi-conn advertisement
6 # Copyright (C) 2022 Red Hat, Inc.
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.
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.
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/>.
22 from contextlib import contextmanager
23 from types import ModuleType
26 from iotests import qemu_img_create, qemu_io
29 disk = os.path.join(iotests.test_dir, 'disk')
31 nbd_sock = os.path.join(iotests.sock_dir, 'nbd_sock')
32 nbd_uri = 'nbd+unix:///{}?socket=' + nbd_sock
36 def open_nbd(export_name):
39 h.connect_uri(nbd_uri.format(export_name))
44 class TestNbdMulticonn(iotests.QMPTestCase):
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)
49 self.vm = iotests.VM()
51 self.vm.cmd('blockdev-add', {
54 'file': {'driver': 'file', 'filename': disk}
66 def run_server(self, max_connections=None):
70 'data': {'path': nbd_sock}
73 if max_connections is not None:
74 args['max-connections'] = max_connections
76 self.vm.cmd('nbd-server-start', args)
79 self.vm.cmd('nbd-server-stop')
81 def add_export(self, name, writable=None):
88 if writable is not None:
89 args['writable'] = writable
91 self.vm.cmd('block-export-add', args)
93 def test_default_settings(self):
94 with self.run_server():
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())
102 def test_limited_connections(self):
103 with self.run_server(max_connections=1):
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())
111 def test_parallel_writes(self):
112 with self.run_server():
113 self.add_export('w', writable=True)
115 clients = [nbd.NBD() for _ in range(3)]
117 c.connect_uri(nbd_uri.format('w'))
118 self.assertTrue(c.can_multi_conn())
120 initial_data = clients[0].pread(1024 * 1024, 0)
121 self.assertEqual(initial_data, b'\x01' * 1024 * 1024)
123 updated_data = b'\x03' * 1024 * 1024
124 clients[1].pwrite(updated_data, 0)
126 current_data = clients[0].pread(1024 * 1024, 0)
128 self.assertEqual(updated_data, current_data)
131 clients[i].shutdown()
134 if __name__ == '__main__':
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
141 iotests.main(supported_fmts=['qcow2'])
143 iotests.notrun('Python bindings to libnbd are not installed')