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