xref: /qemu/tests/qemu-iotests/205 (revision 0885f1221e0add5529dada1e7948d2c00189cb8b)
1 #!/usr/bin/env python3
2 # group: rw quick
3 #
4 # Tests for qmp command nbd-server-remove.
5 #
6 # Copyright (c) 2017 Virtuozzo International GmbH
7 #
8 # This program is free software; you can redistribute it and/or modify
9 # it under the terms of the GNU General Public License as published by
10 # the Free Software Foundation; either version 2 of the License, or
11 # (at your option) any later version.
12 #
13 # This program is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16 # GNU General Public License for more details.
17 #
18 # You should have received a copy of the GNU General Public License
19 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
20 #
21 
22 import os
23 import sys
24 import iotests
25 import time
26 from iotests import qemu_img_create, qemu_io, filter_qemu_io, QemuIoInteractive
27 
28 nbd_sock = os.path.join(iotests.sock_dir, 'nbd_sock')
29 nbd_uri = 'nbd+unix:///exp?socket=' + nbd_sock
30 disk = os.path.join(iotests.test_dir, 'disk')
31 
32 
33 class TestNbdServerRemove(iotests.QMPTestCase):
34     def setUp(self):
35         qemu_img_create('-f', iotests.imgfmt, disk, '1M')
36 
37         self.vm = iotests.VM().add_drive(disk)
38         self.vm.launch()
39 
40         address = {
41             'type': 'unix',
42             'data': {
43                 'path': nbd_sock
44             }
45         }
46 
47         self.vm.cmd('nbd-server-start', addr=address)
48         self.vm.cmd('nbd-server-add', device='drive0', name='exp')
49 
50     def tearDown(self):
51         self.vm.shutdown()
52         os.remove(nbd_sock)
53         os.remove(disk)
54 
55     def remove_export(self, name, mode=None):
56         if mode is None:
57             return self.vm.qmp('nbd-server-remove', name=name)
58         else:
59             return self.vm.qmp('nbd-server-remove', name=name, mode=mode)
60 
61     def assertExportNotFound(self, name):
62         result = self.vm.qmp('nbd-server-remove', name=name)
63         self.assert_qmp(result, 'error/desc', "Export 'exp' is not found")
64 
65     def assertExistingClients(self, result):
66         self.assert_qmp(result, 'error/desc', "export 'exp' still in use")
67 
68     def assertReadOk(self, qemu_io_output):
69         self.assertEqual(
70                 filter_qemu_io(qemu_io_output).strip(),
71                 'read 512/512 bytes at offset 0\n' +
72                 '512 bytes, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)')
73 
74     def assertReadFailed(self, qemu_io_output):
75         self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
76                          'read failed: Input/output error')
77 
78     def assertConnectFailed(self, qemu_io_output):
79         self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
80                          "qemu-io: can't open device " + nbd_uri +
81                          ": Requested export not available\n"
82                          "server reported: export 'exp' not present")
83 
84     def do_test_connect_after_remove(self, mode=None):
85         args = ('-r', '-f', 'raw', '-c', 'read 0 512', nbd_uri)
86         self.assertReadOk(qemu_io(*args).stdout)
87 
88         result = self.remove_export('exp', mode)
89         self.assert_qmp(result, 'return', {})
90 
91         self.assertExportNotFound('exp')
92         self.assertConnectFailed(qemu_io(*args, check=False).stdout)
93 
94     def test_connect_after_remove_default(self):
95         self.do_test_connect_after_remove()
96 
97     def test_connect_after_remove_safe(self):
98         self.do_test_connect_after_remove('safe')
99 
100     def test_connect_after_remove_force(self):
101         self.do_test_connect_after_remove('hard')
102 
103     def do_test_remove_during_connect_safe(self, mode=None):
104         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
105         self.assertReadOk(qio.cmd('read 0 512'))
106 
107         result = self.remove_export('exp', mode)
108         self.assertExistingClients(result)
109 
110         self.assertReadOk(qio.cmd('read 0 512'))
111 
112         qio.close()
113 
114         result = self.remove_export('exp', mode)
115         self.assert_qmp(result, 'return', {})
116 
117         self.assertExportNotFound('exp')
118 
119     def test_remove_during_connect_default(self):
120         self.do_test_remove_during_connect_safe()
121 
122     def test_remove_during_connect_safe(self):
123         self.do_test_remove_during_connect_safe('safe')
124 
125     def test_remove_during_connect_hard(self):
126         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
127         self.assertReadOk(qio.cmd('read 0 512'))
128 
129         result = self.remove_export('exp', 'hard')
130         self.assert_qmp(result, 'return', {})
131 
132         self.assertReadFailed(qio.cmd('read 0 512'))
133         self.assertExportNotFound('exp')
134 
135         qio.close()
136 
137     def test_remove_during_connect_safe_hard(self):
138         qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
139         self.assertReadOk(qio.cmd('read 0 512'))
140 
141         result = self.remove_export('exp', 'safe')
142         self.assertExistingClients(result)
143 
144         self.assertReadOk(qio.cmd('read 0 512'))
145 
146         result = self.remove_export('exp', 'hard')
147         self.assert_qmp(result, 'return', {})
148 
149         self.assertExportNotFound('exp')
150         self.assertReadFailed(qio.cmd('read 0 512'))
151         qio.close()
152 
153 
154 if __name__ == '__main__':
155     iotests.main(supported_fmts=['raw'],
156                  supported_protocols=['nbd'])
157