| #!/usr/bin/env python |
| # |
| # Test case for NBD's blockdev-add interface |
| # |
| # Copyright (C) 2016 Red Hat, Inc. |
| # |
| # This program is free software; you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License as published by |
| # the Free Software Foundation; either version 2 of the License, or |
| # (at your option) any later version. |
| # |
| # This program is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| # GNU General Public License for more details. |
| # |
| # You should have received a copy of the GNU General Public License |
| # along with this program. If not, see <http://www.gnu.org/licenses/>. |
| # |
| |
| import os |
| import socket |
| import stat |
| import time |
| import iotests |
| from iotests import cachemode, imgfmt, qemu_img, qemu_nbd |
| |
| NBD_PORT = 10811 |
| |
| test_img = os.path.join(iotests.test_dir, 'test.img') |
| unix_socket = os.path.join(iotests.test_dir, 'nbd.socket') |
| |
| |
| def flatten_sock_addr(crumpled_address): |
| result = { 'type': crumpled_address['type'] } |
| result.update(crumpled_address['data']) |
| return result |
| |
| |
| class NBDBlockdevAddBase(iotests.QMPTestCase): |
| def blockdev_add_options(self, address, export, node_name): |
| options = { 'node-name': node_name, |
| 'driver': 'raw', |
| 'file': { |
| 'driver': 'nbd', |
| 'read-only': True, |
| 'server': address |
| } } |
| if export is not None: |
| options['file']['export'] = export |
| return options |
| |
| def client_test(self, filename, address, export=None, |
| node_name='nbd-blockdev', delete=True): |
| bao = self.blockdev_add_options(address, export, node_name) |
| result = self.vm.qmp('blockdev-add', **bao) |
| self.assert_qmp(result, 'return', {}) |
| |
| found = False |
| result = self.vm.qmp('query-named-block-nodes') |
| for node in result['return']: |
| if node['node-name'] == node_name: |
| found = True |
| if isinstance(filename, str): |
| self.assert_qmp(node, 'image/filename', filename) |
| else: |
| self.assert_json_filename_equal(node['image']['filename'], |
| filename) |
| break |
| self.assertTrue(found) |
| |
| if delete: |
| result = self.vm.qmp('blockdev-del', node_name=node_name) |
| self.assert_qmp(result, 'return', {}) |
| |
| |
| class QemuNBD(NBDBlockdevAddBase): |
| def setUp(self): |
| qemu_img('create', '-f', iotests.imgfmt, test_img, '64k') |
| self.vm = iotests.VM() |
| self.vm.launch() |
| |
| def tearDown(self): |
| self.vm.shutdown() |
| os.remove(test_img) |
| try: |
| os.remove(unix_socket) |
| except OSError: |
| pass |
| |
| def _server_up(self, *args): |
| self.assertEqual(qemu_nbd('-f', imgfmt, test_img, *args), 0) |
| |
| def test_inet(self): |
| self._server_up('-p', str(NBD_PORT)) |
| address = { 'type': 'inet', |
| 'data': { |
| 'host': 'localhost', |
| 'port': str(NBD_PORT) |
| } } |
| self.client_test('nbd://localhost:%i' % NBD_PORT, |
| flatten_sock_addr(address)) |
| |
| def test_unix(self): |
| self._server_up('-k', unix_socket) |
| address = { 'type': 'unix', |
| 'data': { 'path': unix_socket } } |
| self.client_test('nbd+unix://?socket=' + unix_socket, |
| flatten_sock_addr(address)) |
| |
| |
| class BuiltinNBD(NBDBlockdevAddBase): |
| def setUp(self): |
| qemu_img('create', '-f', iotests.imgfmt, test_img, '64k') |
| self.vm = iotests.VM() |
| self.vm.launch() |
| self.server = iotests.VM('.server') |
| self.server.add_drive_raw('if=none,id=nbd-export,' + |
| 'file=%s,' % test_img + |
| 'format=%s,' % imgfmt + |
| 'cache=%s' % cachemode) |
| self.server.launch() |
| |
| def tearDown(self): |
| self.vm.shutdown() |
| self.server.shutdown() |
| os.remove(test_img) |
| try: |
| os.remove(unix_socket) |
| except OSError: |
| pass |
| |
| def _server_up(self, address, export_name=None, export_name2=None): |
| result = self.server.qmp('nbd-server-start', addr=address) |
| self.assert_qmp(result, 'return', {}) |
| |
| if export_name is None: |
| result = self.server.qmp('nbd-server-add', device='nbd-export') |
| else: |
| result = self.server.qmp('nbd-server-add', device='nbd-export', |
| name=export_name) |
| self.assert_qmp(result, 'return', {}) |
| |
| if export_name2 is not None: |
| result = self.server.qmp('nbd-server-add', device='nbd-export', |
| name=export_name2) |
| self.assert_qmp(result, 'return', {}) |
| |
| |
| def _server_down(self): |
| result = self.server.qmp('nbd-server-stop') |
| self.assert_qmp(result, 'return', {}) |
| |
| def do_test_inet(self, export_name=None): |
| address = { 'type': 'inet', |
| 'data': { |
| 'host': 'localhost', |
| 'port': str(NBD_PORT) |
| } } |
| self._server_up(address, export_name) |
| export_name = export_name or 'nbd-export' |
| self.client_test('nbd://localhost:%i/%s' % (NBD_PORT, export_name), |
| flatten_sock_addr(address), export_name) |
| self._server_down() |
| |
| def test_inet_default_export_name(self): |
| self.do_test_inet() |
| |
| def test_inet_same_export_name(self): |
| self.do_test_inet('nbd-export') |
| |
| def test_inet_different_export_name(self): |
| self.do_test_inet('shadow') |
| |
| def test_inet_two_exports(self): |
| address = { 'type': 'inet', |
| 'data': { |
| 'host': 'localhost', |
| 'port': str(NBD_PORT) |
| } } |
| self._server_up(address, 'exp1', 'exp2') |
| self.client_test('nbd://localhost:%i/%s' % (NBD_PORT, 'exp1'), |
| flatten_sock_addr(address), 'exp1', 'node1', False) |
| self.client_test('nbd://localhost:%i/%s' % (NBD_PORT, 'exp2'), |
| flatten_sock_addr(address), 'exp2', 'node2', False) |
| result = self.vm.qmp('blockdev-del', node_name='node1') |
| self.assert_qmp(result, 'return', {}) |
| result = self.vm.qmp('blockdev-del', node_name='node2') |
| self.assert_qmp(result, 'return', {}) |
| self._server_down() |
| |
| def test_inet6(self): |
| try: |
| socket.getaddrinfo("::0", "0", socket.AF_INET6, |
| socket.SOCK_STREAM, socket.IPPROTO_TCP, |
| socket.AI_ADDRCONFIG | socket.AI_CANONNAME) |
| except socket.gaierror: |
| # IPv6 not available, skip |
| return |
| address = { 'type': 'inet', |
| 'data': { |
| 'host': '::1', |
| 'port': str(NBD_PORT), |
| 'ipv4': False, |
| 'ipv6': True |
| } } |
| filename = { 'driver': 'raw', |
| 'file': { |
| 'driver': 'nbd', |
| 'export': 'nbd-export', |
| 'server': flatten_sock_addr(address) |
| } } |
| self._server_up(address) |
| self.client_test(filename, flatten_sock_addr(address), 'nbd-export') |
| self._server_down() |
| |
| def test_unix(self): |
| address = { 'type': 'unix', |
| 'data': { 'path': unix_socket } } |
| self._server_up(address) |
| self.client_test('nbd+unix:///nbd-export?socket=' + unix_socket, |
| flatten_sock_addr(address), 'nbd-export') |
| self._server_down() |
| |
| def test_fd(self): |
| self._server_up({ 'type': 'unix', |
| 'data': { 'path': unix_socket } }) |
| |
| sockfd = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) |
| sockfd.connect(unix_socket) |
| |
| result = self.vm.send_fd_scm(str(sockfd.fileno())) |
| self.assertEqual(result, 0, 'Failed to send socket FD') |
| |
| result = self.vm.qmp('getfd', fdname='nbd-fifo') |
| self.assert_qmp(result, 'return', {}) |
| |
| address = { 'type': 'fd', |
| 'data': { 'str': 'nbd-fifo' } } |
| filename = { 'driver': 'raw', |
| 'file': { |
| 'driver': 'nbd', |
| 'export': 'nbd-export', |
| 'server': flatten_sock_addr(address) |
| } } |
| self.client_test(filename, flatten_sock_addr(address), 'nbd-export') |
| |
| self._server_down() |
| |
| |
| if __name__ == '__main__': |
| # Need to support image creation |
| iotests.main(supported_fmts=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2', |
| 'vmdk', 'raw', 'vhdx', 'qed']) |
