3 # Test case for NBD's blockdev-add interface
5 # Copyright (C) 2016 Red Hat, Inc.
7 # This program is free software; you can redistribute it and/or modify
8 # it under the terms of the GNU General Public License as published by
9 # the Free Software Foundation; either version 2 of the License, or
10 # (at your option) any later version.
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU General Public License for more details.
17 # You should have received a copy of the GNU General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
27 from iotests
import cachemode
, imgfmt
, qemu_img
, qemu_nbd
, qemu_nbd_pipe
29 NBD_PORT_START
= 32768
30 NBD_PORT_END
= NBD_PORT_START
+ 1024
31 NBD_IPV6_PORT_START
= NBD_PORT_END
32 NBD_IPV6_PORT_END
= NBD_IPV6_PORT_START
+ 1024
34 test_img
= os
.path
.join(iotests
.test_dir
, 'test.img')
35 unix_socket
= os
.path
.join(iotests
.test_dir
, 'nbd.socket')
38 def flatten_sock_addr(crumpled_address
):
39 result
= { 'type': crumpled_address
['type'] }
40 result
.update(crumpled_address
['data'])
44 class NBDBlockdevAddBase(iotests
.QMPTestCase
):
45 def blockdev_add_options(self
, address
, export
, node_name
):
46 options
= { 'node-name': node_name
,
53 if export
is not None:
54 options
['file']['export'] = export
57 def client_test(self
, filename
, address
, export
=None,
58 node_name
='nbd-blockdev', delete
=True):
59 bao
= self
.blockdev_add_options(address
, export
, node_name
)
60 result
= self
.vm
.qmp('blockdev-add', **bao
)
61 self
.assert_qmp(result
, 'return', {})
64 result
= self
.vm
.qmp('query-named-block-nodes')
65 for node
in result
['return']:
66 if node
['node-name'] == node_name
:
68 if isinstance(filename
, str):
69 self
.assert_qmp(node
, 'image/filename', filename
)
71 self
.assert_json_filename_equal(node
['image']['filename'],
74 self
.assertTrue(found
)
77 result
= self
.vm
.qmp('blockdev-del', node_name
=node_name
)
78 self
.assert_qmp(result
, 'return', {})
81 class QemuNBD(NBDBlockdevAddBase
):
83 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, '64k')
84 self
.vm
= iotests
.VM()
91 os
.remove(unix_socket
)
95 def _try_server_up(self
, *args
):
96 status
, msg
= qemu_nbd_pipe('-f', imgfmt
, test_img
, *args
)
99 if 'Address already in use' in msg
:
103 def _server_up(self
, *args
):
104 self
.assertTrue(self
._try
_server
_up
(*args
))
108 nbd_port
= random
.randrange(NBD_PORT_START
, NBD_PORT_END
)
109 if self
._try
_server
_up
('-b', 'localhost', '-p', str(nbd_port
)):
112 address
= { 'type': 'inet',
115 'port': str(nbd_port
)
117 self
.client_test('nbd://localhost:%i' % nbd_port
,
118 flatten_sock_addr(address
))
121 self
._server
_up
('-k', unix_socket
)
122 address
= { 'type': 'unix',
123 'data': { 'path': unix_socket
} }
124 self
.client_test('nbd+unix://?socket=' + unix_socket
,
125 flatten_sock_addr(address
))
128 class BuiltinNBD(NBDBlockdevAddBase
):
130 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, '64k')
131 self
.vm
= iotests
.VM()
133 self
.server
= iotests
.VM('.server')
134 self
.server
.add_drive_raw('if=none,id=nbd-export,' +
135 'file=%s,' % test_img
+
136 'format=%s,' % imgfmt
+
137 'cache=%s' % cachemode
)
142 self
.server
.shutdown()
145 os
.remove(unix_socket
)
149 # Returns False on EADDRINUSE; fails an assertion on other errors.
150 # Returns True on success.
151 def _try_server_up(self
, address
, export_name
=None, export_name2
=None):
152 result
= self
.server
.qmp('nbd-server-start', addr
=address
)
153 if 'error' in result
and \
154 'Address already in use' in result
['error']['desc']:
156 self
.assert_qmp(result
, 'return', {})
158 if export_name
is None:
159 result
= self
.server
.qmp('nbd-server-add', device
='nbd-export')
161 result
= self
.server
.qmp('nbd-server-add', device
='nbd-export',
163 self
.assert_qmp(result
, 'return', {})
165 if export_name2
is not None:
166 result
= self
.server
.qmp('nbd-server-add', device
='nbd-export',
168 self
.assert_qmp(result
, 'return', {})
172 def _server_up(self
, address
, export_name
=None, export_name2
=None):
173 self
.assertTrue(self
._try
_server
_up
(address
, export_name
, export_name2
))
175 def _server_down(self
):
176 result
= self
.server
.qmp('nbd-server-stop')
177 self
.assert_qmp(result
, 'return', {})
179 def do_test_inet(self
, export_name
=None):
181 nbd_port
= random
.randrange(NBD_PORT_START
, NBD_PORT_END
)
182 address
= { 'type': 'inet',
185 'port': str(nbd_port
)
187 if self
._try
_server
_up
(address
, export_name
):
190 export_name
= export_name
or 'nbd-export'
191 self
.client_test('nbd://localhost:%i/%s' % (nbd_port
, export_name
),
192 flatten_sock_addr(address
), export_name
)
195 def test_inet_default_export_name(self
):
198 def test_inet_same_export_name(self
):
199 self
.do_test_inet('nbd-export')
201 def test_inet_different_export_name(self
):
202 self
.do_test_inet('shadow')
204 def test_inet_two_exports(self
):
206 nbd_port
= random
.randrange(NBD_PORT_START
, NBD_PORT_END
)
207 address
= { 'type': 'inet',
210 'port': str(nbd_port
)
212 if self
._try
_server
_up
(address
, 'exp1', 'exp2'):
215 self
.client_test('nbd://localhost:%i/%s' % (nbd_port
, 'exp1'),
216 flatten_sock_addr(address
), 'exp1', 'node1', False)
217 self
.client_test('nbd://localhost:%i/%s' % (nbd_port
, 'exp2'),
218 flatten_sock_addr(address
), 'exp2', 'node2', False)
219 result
= self
.vm
.qmp('blockdev-del', node_name
='node1')
220 self
.assert_qmp(result
, 'return', {})
221 result
= self
.vm
.qmp('blockdev-del', node_name
='node2')
222 self
.assert_qmp(result
, 'return', {})
225 def test_inet6(self
):
227 socket
.getaddrinfo("::0", "0", socket
.AF_INET6
,
228 socket
.SOCK_STREAM
, socket
.IPPROTO_TCP
,
229 socket
.AI_ADDRCONFIG | socket
.AI_CANONNAME
)
230 except socket
.gaierror
:
231 # IPv6 not available, skip
235 nbd_port
= random
.randrange(NBD_IPV6_PORT_START
, NBD_IPV6_PORT_END
)
236 address
= { 'type': 'inet',
239 'port': str(nbd_port
),
243 if self
._try
_server
_up
(address
):
246 filename
= { 'driver': 'raw',
249 'export': 'nbd-export',
250 'server': flatten_sock_addr(address
)
252 self
.client_test(filename
, flatten_sock_addr(address
), 'nbd-export')
256 address
= { 'type': 'unix',
257 'data': { 'path': unix_socket
} }
258 self
._server
_up
(address
)
259 self
.client_test('nbd+unix:///nbd-export?socket=' + unix_socket
,
260 flatten_sock_addr(address
), 'nbd-export')
264 self
._server
_up
({ 'type': 'unix',
265 'data': { 'path': unix_socket
} })
267 sockfd
= socket
.socket(socket
.AF_UNIX
, socket
.SOCK_STREAM
)
268 sockfd
.connect(unix_socket
)
270 result
= self
.vm
.send_fd_scm(fd
=sockfd
.fileno())
271 self
.assertEqual(result
, 0, 'Failed to send socket FD')
273 result
= self
.vm
.qmp('getfd', fdname
='nbd-fifo')
274 self
.assert_qmp(result
, 'return', {})
276 address
= { 'type': 'fd',
277 'data': { 'str': 'nbd-fifo' } }
278 filename
= { 'driver': 'raw',
281 'export': 'nbd-export',
282 'server': flatten_sock_addr(address
)
284 self
.client_test(filename
, flatten_sock_addr(address
), 'nbd-export')
289 if __name__
== '__main__':
290 # Need to support image creation
291 iotests
.main(supported_fmts
=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2',
292 'vmdk', 'raw', 'vhdx', 'qed'])