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/>.
26 from iotests
import cachemode
, imgfmt
, qemu_img
, qemu_nbd
30 test_img
= os
.path
.join(iotests
.test_dir
, 'test.img')
31 unix_socket
= os
.path
.join(iotests
.test_dir
, 'nbd.socket')
34 def flatten_sock_addr(crumpled_address
):
35 result
= { 'type': crumpled_address
['type'] }
36 result
.update(crumpled_address
['data'])
40 class NBDBlockdevAddBase(iotests
.QMPTestCase
):
41 def blockdev_add_options(self
, address
, export
=None):
42 options
= { 'node-name': 'nbd-blockdev',
48 if export
is not None:
49 options
['file']['export'] = export
52 def client_test(self
, filename
, address
, export
=None):
53 bao
= self
.blockdev_add_options(address
, export
)
54 result
= self
.vm
.qmp('blockdev-add', **bao
)
55 self
.assert_qmp(result
, 'return', {})
57 result
= self
.vm
.qmp('query-named-block-nodes')
58 for node
in result
['return']:
59 if node
['node-name'] == 'nbd-blockdev':
60 if isinstance(filename
, str):
61 self
.assert_qmp(node
, 'image/filename', filename
)
63 self
.assert_json_filename_equal(node
['image']['filename'],
67 result
= self
.vm
.qmp('blockdev-del', node_name
='nbd-blockdev')
68 self
.assert_qmp(result
, 'return', {})
71 class QemuNBD(NBDBlockdevAddBase
):
73 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, '64k')
74 self
.vm
= iotests
.VM()
81 os
.remove(unix_socket
)
85 def _server_up(self
, *args
):
86 self
.assertEqual(qemu_nbd('-f', imgfmt
, test_img
, *args
), 0)
89 self
._server
_up
('-p', str(NBD_PORT
))
90 address
= { 'type': 'inet',
95 self
.client_test('nbd://localhost:%i' % NBD_PORT
,
96 flatten_sock_addr(address
))
99 self
._server
_up
('-k', unix_socket
)
100 address
= { 'type': 'unix',
101 'data': { 'path': unix_socket
} }
102 self
.client_test('nbd+unix://?socket=' + unix_socket
,
103 flatten_sock_addr(address
))
106 class BuiltinNBD(NBDBlockdevAddBase
):
108 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, '64k')
109 self
.vm
= iotests
.VM()
111 self
.server
= iotests
.VM('.server')
112 self
.server
.add_drive_raw('if=none,id=nbd-export,' +
113 'file=%s,' % test_img
+
114 'format=%s,' % imgfmt
+
115 'cache=%s' % cachemode
)
120 self
.server
.shutdown()
123 os
.remove(unix_socket
)
127 def _server_up(self
, address
):
128 result
= self
.server
.qmp('nbd-server-start', addr
=address
)
129 self
.assert_qmp(result
, 'return', {})
131 result
= self
.server
.qmp('nbd-server-add', device
='nbd-export')
132 self
.assert_qmp(result
, 'return', {})
134 def _server_down(self
):
135 result
= self
.server
.qmp('nbd-server-stop')
136 self
.assert_qmp(result
, 'return', {})
139 address
= { 'type': 'inet',
142 'port': str(NBD_PORT
)
144 self
._server
_up
(address
)
145 self
.client_test('nbd://localhost:%i/nbd-export' % NBD_PORT
,
146 flatten_sock_addr(address
), 'nbd-export')
149 def test_inet6(self
):
150 address
= { 'type': 'inet',
153 'port': str(NBD_PORT
),
157 filename
= { 'driver': 'raw',
160 'export': 'nbd-export',
161 'server': flatten_sock_addr(address
)
163 self
._server
_up
(address
)
164 self
.client_test(filename
, flatten_sock_addr(address
), 'nbd-export')
168 address
= { 'type': 'unix',
169 'data': { 'path': unix_socket
} }
170 self
._server
_up
(address
)
171 self
.client_test('nbd+unix:///nbd-export?socket=' + unix_socket
,
172 flatten_sock_addr(address
), 'nbd-export')
176 self
._server
_up
({ 'type': 'unix',
177 'data': { 'path': unix_socket
} })
179 sockfd
= socket
.socket(socket
.AF_UNIX
, socket
.SOCK_STREAM
)
180 sockfd
.connect(unix_socket
)
182 result
= self
.vm
.send_fd_scm(str(sockfd
.fileno()))
183 self
.assertEqual(result
, 0, 'Failed to send socket FD')
185 result
= self
.vm
.qmp('getfd', fdname
='nbd-fifo')
186 self
.assert_qmp(result
, 'return', {})
188 address
= { 'type': 'fd',
189 'data': { 'str': 'nbd-fifo' } }
190 filename
= { 'driver': 'raw',
193 'export': 'nbd-export',
194 'server': flatten_sock_addr(address
)
196 self
.client_test(filename
, flatten_sock_addr(address
), 'nbd-export')
201 if __name__
== '__main__':
202 # Need to support image creation
203 iotests
.main(supported_fmts
=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2',
204 'vmdk', 'raw', 'vhdx', 'qed'])