xref: /qemu/tests/qemu-iotests/147 (revision d24eb059faf321fb552cf96a2a1e88e5651347c1) 
1#!/usr/bin/env python3
2# group: img
3#
4# Test case for NBD's blockdev-add interface
5#
6# Copyright (C) 2016 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
22import os
23import random
24import socket
25import stat
26import time
27import iotests
28from iotests import cachemode, aiomode, imgfmt, qemu_img, qemu_nbd, qemu_nbd_early_pipe
29
30NBD_PORT_START      = 32768
31NBD_PORT_END        = NBD_PORT_START + 1024
32NBD_IPV6_PORT_START = NBD_PORT_END
33NBD_IPV6_PORT_END   = NBD_IPV6_PORT_START + 1024
34
35test_img = os.path.join(iotests.test_dir, 'test.img')
36unix_socket = os.path.join(iotests.sock_dir, 'nbd.socket')
37
38
39def flatten_sock_addr(crumpled_address):
40    result = { 'type': crumpled_address['type'] }
41    result.update(crumpled_address['data'])
42    return result
43
44
45class NBDBlockdevAddBase(iotests.QMPTestCase):
46    def blockdev_add_options(self, address, export, node_name):
47        options = { 'node-name': node_name,
48                    'driver': 'raw',
49                    'file': {
50                        'driver': 'nbd',
51                        'read-only': True,
52                        'server': address
53                    } }
54        if export is not None:
55            options['file']['export'] = export
56        return options
57
58    def client_test(self, filename, address, export=None,
59                    node_name='nbd-blockdev', delete=True):
60        bao = self.blockdev_add_options(address, export, node_name)
61        result = self.vm.qmp('blockdev-add', **bao)
62        self.assert_qmp(result, 'return', {})
63
64        found = False
65        result = self.vm.qmp('query-named-block-nodes')
66        for node in result['return']:
67            if node['node-name'] == node_name:
68                found = True
69                if isinstance(filename, str):
70                    self.assert_qmp(node, 'image/filename', filename)
71                else:
72                    self.assert_json_filename_equal(node['image']['filename'],
73                                                    filename)
74                break
75        self.assertTrue(found)
76
77        if delete:
78            result = self.vm.qmp('blockdev-del', node_name=node_name)
79            self.assert_qmp(result, 'return', {})
80
81
82class QemuNBD(NBDBlockdevAddBase):
83    def setUp(self):
84        qemu_img('create', '-f', iotests.imgfmt, test_img, '64k')
85        self.vm = iotests.VM()
86        self.vm.launch()
87
88    def tearDown(self):
89        self.vm.shutdown()
90        os.remove(test_img)
91        try:
92            os.remove(unix_socket)
93        except OSError:
94            pass
95
96    def _try_server_up(self, *args):
97        status, msg = qemu_nbd_early_pipe('-f', imgfmt, test_img, *args)
98        if status == 0:
99            return True
100        if 'Address already in use' in msg:
101            return False
102        self.fail(msg)
103
104    def _server_up(self, *args):
105        self.assertTrue(self._try_server_up(*args))
106
107    def test_inet(self):
108        while True:
109            nbd_port = random.randrange(NBD_PORT_START, NBD_PORT_END)
110            if self._try_server_up('-b', 'localhost', '-p', str(nbd_port)):
111                break
112
113        address = { 'type': 'inet',
114                    'data': {
115                        'host': 'localhost',
116                        'port': str(nbd_port)
117                    } }
118        self.client_test('nbd://localhost:%i' % nbd_port,
119                         flatten_sock_addr(address))
120
121    def test_unix(self):
122        self._server_up('-k', unix_socket)
123        address = { 'type': 'unix',
124                    'data': { 'path': unix_socket } }
125        self.client_test('nbd+unix://?socket=' + unix_socket,
126                         flatten_sock_addr(address))
127
128
129class BuiltinNBD(NBDBlockdevAddBase):
130    def setUp(self):
131        qemu_img('create', '-f', iotests.imgfmt, test_img, '64k')
132        self.vm = iotests.VM()
133        self.vm.launch()
134        self.server = iotests.VM('.server')
135        self.server.add_drive_raw('if=none,id=nbd-export,' +
136                                  'file=%s,' % test_img +
137                                  'format=%s,' % imgfmt +
138                                  'cache=%s,' % cachemode +
139                                  'aio=%s' % aiomode)
140        self.server.launch()
141
142    def tearDown(self):
143        self.vm.shutdown()
144        self.server.shutdown()
145        os.remove(test_img)
146        try:
147            os.remove(unix_socket)
148        except OSError:
149            pass
150
151    # Returns False on EADDRINUSE; fails an assertion on other errors.
152    # Returns True on success.
153    def _try_server_up(self, address, export_name=None, export_name2=None):
154        result = self.server.qmp('nbd-server-start', addr=address)
155        if 'error' in result and \
156           'Address already in use' in result['error']['desc']:
157            return False
158        self.assert_qmp(result, 'return', {})
159
160        if export_name is None:
161            result = self.server.qmp('nbd-server-add', device='nbd-export')
162            self.assert_qmp(result, 'return', {})
163        else:
164            result = self.server.qmp('nbd-server-add', device='nbd-export',
165                                     name=export_name)
166            self.assert_qmp(result, 'return', {})
167
168        if export_name2 is not None:
169            result = self.server.qmp('nbd-server-add', device='nbd-export',
170                                     name=export_name2)
171            self.assert_qmp(result, 'return', {})
172
173        return True
174
175    def _server_up(self, address, export_name=None, export_name2=None):
176        self.assertTrue(self._try_server_up(address, export_name, export_name2))
177
178    def _server_down(self):
179        result = self.server.qmp('nbd-server-stop')
180        self.assert_qmp(result, 'return', {})
181
182    def do_test_inet(self, export_name=None):
183        while True:
184            nbd_port = random.randrange(NBD_PORT_START, NBD_PORT_END)
185            address = { 'type': 'inet',
186                        'data': {
187                            'host': 'localhost',
188                            'port': str(nbd_port)
189                        } }
190            if self._try_server_up(address, export_name):
191                break
192
193        export_name = export_name or 'nbd-export'
194        self.client_test('nbd://localhost:%i/%s' % (nbd_port, export_name),
195                         flatten_sock_addr(address), export_name)
196        self._server_down()
197
198    def test_inet_default_export_name(self):
199        self.do_test_inet()
200
201    def test_inet_same_export_name(self):
202        self.do_test_inet('nbd-export')
203
204    def test_inet_different_export_name(self):
205        self.do_test_inet('shadow')
206
207    def test_inet_two_exports(self):
208        while True:
209            nbd_port = random.randrange(NBD_PORT_START, NBD_PORT_END)
210            address = { 'type': 'inet',
211                        'data': {
212                            'host': 'localhost',
213                            'port': str(nbd_port)
214                        } }
215            if self._try_server_up(address, 'exp1', 'exp2'):
216                break
217
218        self.client_test('nbd://localhost:%i/%s' % (nbd_port, 'exp1'),
219                         flatten_sock_addr(address), 'exp1', 'node1', False)
220        self.client_test('nbd://localhost:%i/%s' % (nbd_port, 'exp2'),
221                         flatten_sock_addr(address), 'exp2', 'node2', False)
222        result = self.vm.qmp('blockdev-del', node_name='node1')
223        self.assert_qmp(result, 'return', {})
224        result = self.vm.qmp('blockdev-del', node_name='node2')
225        self.assert_qmp(result, 'return', {})
226        self._server_down()
227
228    def test_inet6(self):
229        try:
230            socket.getaddrinfo("::0", "0", socket.AF_INET6,
231                               socket.SOCK_STREAM, socket.IPPROTO_TCP,
232                               socket.AI_ADDRCONFIG | socket.AI_CANONNAME)
233        except socket.gaierror:
234            # IPv6 not available, skip
235            return
236
237        while True:
238            nbd_port = random.randrange(NBD_IPV6_PORT_START, NBD_IPV6_PORT_END)
239            address = { 'type': 'inet',
240                        'data': {
241                            'host': '::1',
242                            'port': str(nbd_port),
243                            'ipv4': False,
244                            'ipv6': True
245                        } }
246            if self._try_server_up(address):
247                break
248
249        filename = { 'driver': 'raw',
250                     'file': {
251                         'driver': 'nbd',
252                         'export': 'nbd-export',
253                         'server': flatten_sock_addr(address)
254                     } }
255        self.client_test(filename, flatten_sock_addr(address), 'nbd-export')
256        self._server_down()
257
258    def test_unix(self):
259        address = { 'type': 'unix',
260                    'data': { 'path': unix_socket } }
261        self._server_up(address)
262        self.client_test('nbd+unix:///nbd-export?socket=' + unix_socket,
263                         flatten_sock_addr(address), 'nbd-export')
264        self._server_down()
265
266    def test_fd(self):
267        self._server_up({ 'type': 'unix',
268                          'data': { 'path': unix_socket } })
269
270        sockfd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
271        sockfd.connect(unix_socket)
272
273        result = self.vm.send_fd_scm(fd=sockfd.fileno())
274        self.assertEqual(result, 0, 'Failed to send socket FD')
275
276        result = self.vm.qmp('getfd', fdname='nbd-fifo')
277        self.assert_qmp(result, 'return', {})
278
279        address = { 'type': 'fd',
280                    'data': { 'str': 'nbd-fifo' } }
281        filename = { 'driver': 'raw',
282                     'file': {
283                         'driver': 'nbd',
284                         'export': 'nbd-export',
285                         'server': flatten_sock_addr(address)
286                     } }
287        self.client_test(filename, flatten_sock_addr(address), 'nbd-export')
288
289        self._server_down()
290
291
292if __name__ == '__main__':
293    iotests.main(supported_fmts=['raw'],
294                 supported_protocols=['nbd'])
295