cirrus: fix off-by-one in cirrus_bitblt_rop_bkwd_transp_*_16
[qemu.git] / 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'])