mirror of
https://github.com/qemu/qemu.git
synced 2025-10-31 04:06:46 +00:00
549084eaed
Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com> Reviewed-by: Eric Blake <eblake@redhat.com> Message-Id: <20180119135719.24745-4-vsementsov@virtuozzo.com> Signed-off-by: Eric Blake <eblake@redhat.com>
255 lines
8.9 KiB
Python
Executable File
255 lines
8.9 KiB
Python
Executable File
#!/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'])
|