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 'read-only': True, 47 'server': address 48 } } 49 if export is not None: 50 options['file']['export'] = export 51 return options 52 53 def client_test(self, filename, address, export=None): 54 bao = self.blockdev_add_options(address, export) 55 result = self.vm.qmp('blockdev-add', **bao) 56 self.assert_qmp(result, 'return', {}) 57 58 result = self.vm.qmp('query-named-block-nodes') 59 for node in result['return']: 60 if node['node-name'] == 'nbd-blockdev': 61 if isinstance(filename, str): 62 self.assert_qmp(node, 'image/filename', filename) 63 else: 64 self.assert_json_filename_equal(node['image']['filename'], 65 filename) 66 break 67 68 result = self.vm.qmp('blockdev-del', node_name='nbd-blockdev') 69 self.assert_qmp(result, 'return', {}) 70 71 72class QemuNBD(NBDBlockdevAddBase): 73 def setUp(self): 74 qemu_img('create', '-f', iotests.imgfmt, test_img, '64k') 75 self.vm = iotests.VM() 76 self.vm.launch() 77 78 def tearDown(self): 79 self.vm.shutdown() 80 os.remove(test_img) 81 try: 82 os.remove(unix_socket) 83 except OSError: 84 pass 85 86 def _server_up(self, *args): 87 self.assertEqual(qemu_nbd('-f', imgfmt, test_img, *args), 0) 88 89 def test_inet(self): 90 self._server_up('-p', str(NBD_PORT)) 91 address = { 'type': 'inet', 92 'data': { 93 'host': 'localhost', 94 'port': str(NBD_PORT) 95 } } 96 self.client_test('nbd://localhost:%i' % NBD_PORT, 97 flatten_sock_addr(address)) 98 99 def test_unix(self): 100 self._server_up('-k', unix_socket) 101 address = { 'type': 'unix', 102 'data': { 'path': unix_socket } } 103 self.client_test('nbd+unix://?socket=' + unix_socket, 104 flatten_sock_addr(address)) 105 106 107class BuiltinNBD(NBDBlockdevAddBase): 108 def setUp(self): 109 qemu_img('create', '-f', iotests.imgfmt, test_img, '64k') 110 self.vm = iotests.VM() 111 self.vm.launch() 112 self.server = iotests.VM('.server') 113 self.server.add_drive_raw('if=none,id=nbd-export,' + 114 'file=%s,' % test_img + 115 'format=%s,' % imgfmt + 116 'cache=%s' % cachemode) 117 self.server.launch() 118 119 def tearDown(self): 120 self.vm.shutdown() 121 self.server.shutdown() 122 os.remove(test_img) 123 try: 124 os.remove(unix_socket) 125 except OSError: 126 pass 127 128 def _server_up(self, address): 129 result = self.server.qmp('nbd-server-start', addr=address) 130 self.assert_qmp(result, 'return', {}) 131 132 result = self.server.qmp('nbd-server-add', device='nbd-export') 133 self.assert_qmp(result, 'return', {}) 134 135 def _server_down(self): 136 result = self.server.qmp('nbd-server-stop') 137 self.assert_qmp(result, 'return', {}) 138 139 def test_inet(self): 140 address = { 'type': 'inet', 141 'data': { 142 'host': 'localhost', 143 'port': str(NBD_PORT) 144 } } 145 self._server_up(address) 146 self.client_test('nbd://localhost:%i/nbd-export' % NBD_PORT, 147 flatten_sock_addr(address), 'nbd-export') 148 self._server_down() 149 150 def test_inet6(self): 151 try: 152 socket.getaddrinfo("::0", "0", socket.AF_INET6, 153 socket.SOCK_STREAM, socket.IPPROTO_TCP, 154 socket.AI_ADDRCONFIG | socket.AI_CANONNAME) 155 except socket.gaierror: 156 # IPv6 not available, skip 157 return 158 address = { 'type': 'inet', 159 'data': { 160 'host': '::1', 161 'port': str(NBD_PORT), 162 'ipv4': False, 163 'ipv6': True 164 } } 165 filename = { 'driver': 'raw', 166 'file': { 167 'driver': 'nbd', 168 'export': 'nbd-export', 169 'server': flatten_sock_addr(address) 170 } } 171 self._server_up(address) 172 self.client_test(filename, flatten_sock_addr(address), 'nbd-export') 173 self._server_down() 174 175 def test_unix(self): 176 address = { 'type': 'unix', 177 'data': { 'path': unix_socket } } 178 self._server_up(address) 179 self.client_test('nbd+unix:///nbd-export?socket=' + unix_socket, 180 flatten_sock_addr(address), 'nbd-export') 181 self._server_down() 182 183 def test_fd(self): 184 self._server_up({ 'type': 'unix', 185 'data': { 'path': unix_socket } }) 186 187 sockfd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 188 sockfd.connect(unix_socket) 189 190 result = self.vm.send_fd_scm(str(sockfd.fileno())) 191 self.assertEqual(result, 0, 'Failed to send socket FD') 192 193 result = self.vm.qmp('getfd', fdname='nbd-fifo') 194 self.assert_qmp(result, 'return', {}) 195 196 address = { 'type': 'fd', 197 'data': { 'str': 'nbd-fifo' } } 198 filename = { 'driver': 'raw', 199 'file': { 200 'driver': 'nbd', 201 'export': 'nbd-export', 202 'server': flatten_sock_addr(address) 203 } } 204 self.client_test(filename, flatten_sock_addr(address), 'nbd-export') 205 206 self._server_down() 207 208 209if __name__ == '__main__': 210 # Need to support image creation 211 iotests.main(supported_fmts=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2', 212 'vmdk', 'raw', 'vhdx', 'qed']) 213