xref: /qemu/tests/qemu-iotests/tests/nbd-multiconn (revision b49f4755) 
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
21import os
22from contextlib import contextmanager
23from types import ModuleType
24
25import iotests
26from iotests import qemu_img_create, qemu_io
27
28
29disk = os.path.join(iotests.test_dir, 'disk')
30size = '4M'
31nbd_sock = os.path.join(iotests.sock_dir, 'nbd_sock')
32nbd_uri = 'nbd+unix:///{}?socket=' + nbd_sock
33nbd: ModuleType
34
35@contextmanager
36def 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
44class 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
134if __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