1#!/usr/bin/env python 2# 3# Test case for NBD's blockdev-add interface 4# 5# Copyright (C) 2016 Red Hat, Inc. 6# 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. 11# 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. 16# 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/>. 19# 20 21import os 22import socket 23import stat 24import time 25import iotests 26from iotests import cachemode, imgfmt, qemu_img, qemu_nbd 27 28NBD_PORT = 10811 29 30test_img = os.path.join(iotests.test_dir, 'test.img') 31unix_socket = os.path.join(iotests.test_dir, 'nbd.socket') 32 33 34def flatten_sock_addr(crumpled_address): 35 result = { 'type': crumpled_address['type'] } 36 result.update(crumpled_address['data']) 37 return result 38 39 40class NBDBlockdevAddBase(iotests.QMPTestCase): 41 def blockdev_add_options(self, address, export=None): 42 options = { 'node-name': 'nbd-blockdev', 43 'driver': 'raw', 44 'file': { 45 'driver': 'nbd', 46 'server': address 47 } } 48 if export is not None: 49 options['file']['export'] = export 50 return options 51 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', {}) 56 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) 62 else: 63 self.assert_json_filename_equal(node['image']['filename'], 64 filename) 65 break 66 67 result = self.vm.qmp('blockdev-del', node_name='nbd-blockdev') 68 self.assert_qmp(result, 'return', {}) 69 70 71class QemuNBD(NBDBlockdevAddBase): 72 def setUp(self): 73 qemu_img('create', '-f', iotests.imgfmt, test_img, '64k') 74 self.vm = iotests.VM() 75 self.vm.launch() 76 77 def tearDown(self): 78 self.vm.shutdown() 79 os.remove(test_img) 80 try: 81 os.remove(unix_socket) 82 except OSError: 83 pass 84 85 def _server_up(self, *args): 86 self.assertEqual(qemu_nbd('-f', imgfmt, test_img, *args), 0) 87 88 def test_inet(self): 89 self._server_up('-p', str(NBD_PORT)) 90 address = { 'type': 'inet', 91 'data': { 92 'host': 'localhost', 93 'port': str(NBD_PORT) 94 } } 95 self.client_test('nbd://localhost:%i' % NBD_PORT, 96 flatten_sock_addr(address)) 97 98 def test_unix(self): 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)) 104 105 106class BuiltinNBD(NBDBlockdevAddBase): 107 def setUp(self): 108 qemu_img('create', '-f', iotests.imgfmt, test_img, '64k') 109 self.vm = iotests.VM() 110 self.vm.launch() 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) 116 self.server.launch() 117 118 def tearDown(self): 119 self.vm.shutdown() 120 self.server.shutdown() 121 os.remove(test_img) 122 try: 123 os.remove(unix_socket) 124 except OSError: 125 pass 126 127 def _server_up(self, address): 128 result = self.server.qmp('nbd-server-start', addr=address) 129 self.assert_qmp(result, 'return', {}) 130 131 result = self.server.qmp('nbd-server-add', device='nbd-export') 132 self.assert_qmp(result, 'return', {}) 133 134 def _server_down(self): 135 result = self.server.qmp('nbd-server-stop') 136 self.assert_qmp(result, 'return', {}) 137 138 def test_inet(self): 139 address = { 'type': 'inet', 140 'data': { 141 'host': 'localhost', 142 'port': str(NBD_PORT) 143 } } 144 self._server_up(address) 145 self.client_test('nbd://localhost:%i/nbd-export' % NBD_PORT, 146 flatten_sock_addr(address), 'nbd-export') 147 self._server_down() 148 149 def test_inet6(self): 150 address = { 'type': 'inet', 151 'data': { 152 'host': '::1', 153 'port': str(NBD_PORT), 154 'ipv4': False, 155 'ipv6': True 156 } } 157 filename = { 'driver': 'raw', 158 'file': { 159 'driver': 'nbd', 160 'export': 'nbd-export', 161 'server': flatten_sock_addr(address) 162 } } 163 self._server_up(address) 164 self.client_test(filename, flatten_sock_addr(address), 'nbd-export') 165 self._server_down() 166 167 def test_unix(self): 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') 173 self._server_down() 174 175 def test_fd(self): 176 self._server_up({ 'type': 'unix', 177 'data': { 'path': unix_socket } }) 178 179 sockfd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 180 sockfd.connect(unix_socket) 181 182 result = self.vm.send_fd_scm(str(sockfd.fileno())) 183 self.assertEqual(result, 0, 'Failed to send socket FD') 184 185 result = self.vm.qmp('getfd', fdname='nbd-fifo') 186 self.assert_qmp(result, 'return', {}) 187 188 address = { 'type': 'fd', 189 'data': { 'str': 'nbd-fifo' } } 190 filename = { 'driver': 'raw', 191 'file': { 192 'driver': 'nbd', 193 'export': 'nbd-export', 194 'server': flatten_sock_addr(address) 195 } } 196 self.client_test(filename, flatten_sock_addr(address), 'nbd-export') 197 198 self._server_down() 199 200 201if __name__ == '__main__': 202 # Need to support image creation 203 iotests.main(supported_fmts=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2', 204 'vmdk', 'raw', 'vhdx', 'qed']) 205