tests/qemu-iotests/147

   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
  21 import os
  22 import socket
  23 import stat
  24 import time
  25 import iotests
  26 from iotests import cachemode, imgfmt, qemu_img, qemu_nbd
  27
  28 NBD_PORT = 10811
  29
  30 test_img = os.path.join(iotests.test_dir, 'test.img')
  31 unix_socket = os.path.join(iotests.test_dir, 'nbd.socket')
  32
  33 class 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
  64 class 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
  97 class 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
 192 if __name__ == '__main__':
 193     # Need to support image creation
 194     iotests.main(supported_fmts=['vpc', 'parallels', 'qcow', 'vdi', 'qcow2',
 195                                  'vmdk', 'raw', 'vhdx', 'qed'])