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 try: 151 socket.getaddrinfo("::0", "0", socket.AF_INET6, 152 socket.SOCK_STREAM, socket.IPPROTO_TCP, 153 socket.AI_ADDRCONFIG | socket.AI_CANONNAME) 154 except socket.gaierror: 155 # IPv6 not available, skip 156 return 157 address = { 'type': 'inet', 158 'data': { 159 'host': '::1', 160 'port': str(NBD_PORT), 161 'ipv4': False, 162 'ipv6': True 163 } } 164 filename = { 'driver': 'raw', 165 'file': { 166 'driver': 'nbd', 167 'export': 'nbd-export', 168 'server': flatten_sock_addr(address) 169 } } 170 self._server_up(address) 171 self.client_test(filename, flatten_sock_addr(address), 'nbd-export') 172 self._server_down() 173 174 def test_unix(self): 175 address = { 'type': 'unix', 176 'data': { 'path': unix_socket } } 177 self._server_up(address) 178 self.client_test('nbd+unix:///nbd-export?socket=' + unix_socket, 179 flatten_sock_addr(address), 'nbd-export') 180 self._server_down() 181 182 def test_fd(self): 183 self._server_up({ 'type': 'unix', 184 'data': { 'path': unix_socket } }) 185 186 sockfd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 187 sockfd.connect(unix_socket) 188 189 result = self.vm.send_fd_scm(str(sockfd.fileno())) 190 self.assertEqual(result, 0, 'Failed to send socket FD') 191 192 result = self.vm.qmp('getfd', fdname='nbd-fifo') 193 self.assert_qmp(result, 'return', {}) 194 195 address = { 'type': 'fd', 196 'data': { 'str': 'nbd-fifo' } } 197 filename = { 'driver': 'raw', 198 'file': { 199 'driver': 'nbd', 200 'export': 'nbd-export', 201 'server': flatten_sock_addr(address) 202 } } 203 self.client_test(filename, flatten_sock_addr(address), 'nbd-export') 204 205 self._server_down() 206 207 208if __name__ == '__main__': 209 # Need to support image creation 210 iotests.main(supported_fmts=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2', 211 'vmdk', 'raw', 'vhdx', 'qed']) 212