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