xref: /openbmc/qemu/tests/qemu-iotests/205 (revision a68694cd) 
1#!/usr/bin/env python3
2#
3# Tests for qmp command nbd-server-remove.
4#
5# Copyright (c) 2017 Virtuozzo International GmbH
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
21import os
22import sys
23import iotests
24import time
25from iotests import qemu_img_create, qemu_io, filter_qemu_io, QemuIoInteractive
26
27nbd_sock = os.path.join(iotests.sock_dir, 'nbd_sock')
28nbd_uri = 'nbd+unix:///exp?socket=' + nbd_sock
29disk = os.path.join(iotests.test_dir, 'disk')
30
31
32class TestNbdServerRemove(iotests.QMPTestCase):
33    def setUp(self):
34        qemu_img_create('-f', iotests.imgfmt, disk, '1M')
35
36        self.vm = iotests.VM().add_drive(disk)
37        self.vm.launch()
38
39        address = {
40            'type': 'unix',
41            'data': {
42                'path': nbd_sock
43            }
44        }
45
46        result = self.vm.qmp('nbd-server-start', addr=address)
47        self.assert_qmp(result, 'return', {})
48        result = self.vm.qmp('nbd-server-add', device='drive0', name='exp')
49        self.assert_qmp(result, 'return', {})
50
51    def tearDown(self):
52        self.vm.shutdown()
53        os.remove(nbd_sock)
54        os.remove(disk)
55
56    def remove_export(self, name, mode=None):
57        if mode is None:
58            return self.vm.qmp('nbd-server-remove', name=name)
59        else:
60            return self.vm.qmp('nbd-server-remove', name=name, mode=mode)
61
62    def assertExportNotFound(self, name):
63        result = self.vm.qmp('nbd-server-remove', name=name)
64        self.assert_qmp(result, 'error/desc', "Export 'exp' is not found")
65
66    def assertExistingClients(self, result):
67        self.assert_qmp(result, 'error/desc', "export 'exp' still in use")
68
69    def assertReadOk(self, qemu_io_output):
70        self.assertEqual(
71                filter_qemu_io(qemu_io_output).strip(),
72                'read 512/512 bytes at offset 0\n' +
73                '512 bytes, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec)')
74
75    def assertReadFailed(self, qemu_io_output):
76        self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
77                         'read failed: Input/output error')
78
79    def assertConnectFailed(self, qemu_io_output):
80        self.assertEqual(filter_qemu_io(qemu_io_output).strip(),
81                         "qemu-io: can't open device " + nbd_uri +
82                         ": Requested export not available\n"
83                         "server reported: export 'exp' not present")
84
85    def do_test_connect_after_remove(self, mode=None):
86        args = ('-r', '-f', 'raw', '-c', 'read 0 512', nbd_uri)
87        self.assertReadOk(qemu_io(*args))
88
89        result = self.remove_export('exp', mode)
90        self.assert_qmp(result, 'return', {})
91
92        self.assertExportNotFound('exp')
93        self.assertConnectFailed(qemu_io(*args))
94
95    def test_connect_after_remove_default(self):
96        self.do_test_connect_after_remove()
97
98    def test_connect_after_remove_safe(self):
99        self.do_test_connect_after_remove('safe')
100
101    def test_connect_after_remove_force(self):
102        self.do_test_connect_after_remove('hard')
103
104    def do_test_remove_during_connect_safe(self, mode=None):
105        qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
106        self.assertReadOk(qio.cmd('read 0 512'))
107
108        result = self.remove_export('exp', mode)
109        self.assertExistingClients(result)
110
111        self.assertReadOk(qio.cmd('read 0 512'))
112
113        qio.close()
114
115        result = self.remove_export('exp', mode)
116        self.assert_qmp(result, 'return', {})
117
118        self.assertExportNotFound('exp')
119
120    def test_remove_during_connect_default(self):
121        self.do_test_remove_during_connect_safe()
122
123    def test_remove_during_connect_safe(self):
124        self.do_test_remove_during_connect_safe('safe')
125
126    def test_remove_during_connect_hard(self):
127        qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
128        self.assertReadOk(qio.cmd('read 0 512'))
129
130        result = self.remove_export('exp', 'hard')
131        self.assert_qmp(result, 'return', {})
132
133        self.assertReadFailed(qio.cmd('read 0 512'))
134        self.assertExportNotFound('exp')
135
136        qio.close()
137
138    def test_remove_during_connect_safe_hard(self):
139        qio = QemuIoInteractive('-r', '-f', 'raw', nbd_uri)
140        self.assertReadOk(qio.cmd('read 0 512'))
141
142        result = self.remove_export('exp', 'safe')
143        self.assertExistingClients(result)
144
145        self.assertReadOk(qio.cmd('read 0 512'))
146
147        result = self.remove_export('exp', 'hard')
148        self.assert_qmp(result, 'return', {})
149
150        self.assertExportNotFound('exp')
151        self.assertReadFailed(qio.cmd('read 0 512'))
152        qio.close()
153
154
155if __name__ == '__main__':
156    iotests.main(supported_fmts=['raw'],
157                 supported_protocols=['nbd'])
158