3 # Tests for drive-backup
5 # Copyright (C) 2013 Red Hat, Inc.
9 # This program is free software; you can redistribute it and/or modify
10 # it under the terms of the GNU General Public License as published by
11 # the Free Software Foundation; either version 2 of the License, or
12 # (at your option) any later version.
14 # This program is distributed in the hope that it will be useful,
15 # but WITHOUT ANY WARRANTY; without even the implied warranty of
16 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 # GNU General Public License for more details.
19 # You should have received a copy of the GNU General Public License
20 # along with this program. If not, see <http://www.gnu.org/licenses/>.
26 from iotests
import qemu_img
, qemu_io
28 test_img
= os
.path
.join(iotests
.test_dir
, 'test.img')
29 target_img
= os
.path
.join(iotests
.test_dir
, 'target.img')
31 class TestSingleDrive(iotests
.QMPTestCase
):
32 image_len
= 64 * 1024 * 1024 # MB
35 # Write data to the image so we can compare later
36 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, str(TestSingleDrive
.image_len
))
37 qemu_io('-c', 'write -P0x5d 0 64k', test_img
)
38 qemu_io('-c', 'write -P0xd5 1M 32k', test_img
)
39 qemu_io('-c', 'write -P0xdc 32M 124k', test_img
)
40 qemu_io('-c', 'write -P0xdc 67043328 64k', test_img
)
42 self
.vm
= iotests
.VM().add_drive(test_img
)
53 def test_cancel(self
):
54 self
.assert_no_active_block_jobs()
56 result
= self
.vm
.qmp('drive-backup', device
='drive0',
57 target
=target_img
, sync
='full')
58 self
.assert_qmp(result
, 'return', {})
60 event
= self
.cancel_and_wait()
61 self
.assert_qmp(event
, 'data/type', 'backup')
64 self
.assert_no_active_block_jobs()
66 self
.vm
.pause_drive('drive0')
67 result
= self
.vm
.qmp('drive-backup', device
='drive0',
68 target
=target_img
, sync
='full')
69 self
.assert_qmp(result
, 'return', {})
71 result
= self
.vm
.qmp('block-job-pause', device
='drive0')
72 self
.assert_qmp(result
, 'return', {})
74 self
.vm
.resume_drive('drive0')
76 result
= self
.vm
.qmp('query-block-jobs')
77 offset
= self
.dictpath(result
, 'return[0]/offset')
80 result
= self
.vm
.qmp('query-block-jobs')
81 self
.assert_qmp(result
, 'return[0]/offset', offset
)
83 result
= self
.vm
.qmp('block-job-resume', device
='drive0')
84 self
.assert_qmp(result
, 'return', {})
86 self
.wait_until_completed()
89 self
.assertTrue(iotests
.compare_images(test_img
, target_img
),
90 'target image does not match source after backup')
92 def test_medium_not_found(self
):
93 result
= self
.vm
.qmp('drive-backup', device
='ide1-cd0',
94 target
=target_img
, sync
='full')
95 self
.assert_qmp(result
, 'error/class', 'GenericError')
97 def test_image_not_found(self
):
98 result
= self
.vm
.qmp('drive-backup', device
='drive0',
99 target
=target_img
, sync
='full', mode
='existing')
100 self
.assert_qmp(result
, 'error/class', 'GenericError')
102 def test_invalid_format(self
):
103 result
= self
.vm
.qmp('drive-backup', device
='drive0',
104 target
=target_img
, sync
='full',
105 format
='spaghetti-noodles')
106 self
.assert_qmp(result
, 'error/class', 'GenericError')
108 def test_device_not_found(self
):
109 result
= self
.vm
.qmp('drive-backup', device
='nonexistent',
110 target
=target_img
, sync
='full')
111 self
.assert_qmp(result
, 'error/class', 'DeviceNotFound')
113 class TestSetSpeed(iotests
.QMPTestCase
):
114 image_len
= 80 * 1024 * 1024 # MB
117 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, str(TestSetSpeed
.image_len
))
118 qemu_io('-c', 'write -P1 0 512', test_img
)
119 self
.vm
= iotests
.VM().add_drive(test_img
)
125 os
.remove(target_img
)
127 def test_set_speed(self
):
128 self
.assert_no_active_block_jobs()
130 self
.vm
.pause_drive('drive0')
131 result
= self
.vm
.qmp('drive-backup', device
='drive0',
132 target
=target_img
, sync
='full')
133 self
.assert_qmp(result
, 'return', {})
136 result
= self
.vm
.qmp('query-block-jobs')
137 self
.assert_qmp(result
, 'return[0]/device', 'drive0')
138 self
.assert_qmp(result
, 'return[0]/speed', 0)
140 result
= self
.vm
.qmp('block-job-set-speed', device
='drive0', speed
=8 * 1024 * 1024)
141 self
.assert_qmp(result
, 'return', {})
143 # Ensure the speed we set was accepted
144 result
= self
.vm
.qmp('query-block-jobs')
145 self
.assert_qmp(result
, 'return[0]/device', 'drive0')
146 self
.assert_qmp(result
, 'return[0]/speed', 8 * 1024 * 1024)
148 event
= self
.cancel_and_wait(resume
=True)
149 self
.assert_qmp(event
, 'data/type', 'backup')
151 # Check setting speed in drive-backup works
152 self
.vm
.pause_drive('drive0')
153 result
= self
.vm
.qmp('drive-backup', device
='drive0',
154 target
=target_img
, sync
='full', speed
=4*1024*1024)
155 self
.assert_qmp(result
, 'return', {})
157 result
= self
.vm
.qmp('query-block-jobs')
158 self
.assert_qmp(result
, 'return[0]/device', 'drive0')
159 self
.assert_qmp(result
, 'return[0]/speed', 4 * 1024 * 1024)
161 event
= self
.cancel_and_wait(resume
=True)
162 self
.assert_qmp(event
, 'data/type', 'backup')
164 def test_set_speed_invalid(self
):
165 self
.assert_no_active_block_jobs()
167 result
= self
.vm
.qmp('drive-backup', device
='drive0',
168 target
=target_img
, sync
='full', speed
=-1)
169 self
.assert_qmp(result
, 'error/class', 'GenericError')
171 self
.assert_no_active_block_jobs()
173 self
.vm
.pause_drive('drive0')
174 result
= self
.vm
.qmp('drive-backup', device
='drive0',
175 target
=target_img
, sync
='full')
176 self
.assert_qmp(result
, 'return', {})
178 result
= self
.vm
.qmp('block-job-set-speed', device
='drive0', speed
=-1)
179 self
.assert_qmp(result
, 'error/class', 'GenericError')
181 event
= self
.cancel_and_wait(resume
=True)
182 self
.assert_qmp(event
, 'data/type', 'backup')
184 class TestSingleTransaction(iotests
.QMPTestCase
):
185 image_len
= 64 * 1024 * 1024 # MB
188 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, str(TestSingleTransaction
.image_len
))
189 qemu_io('-c', 'write -P0x5d 0 64k', test_img
)
190 qemu_io('-c', 'write -P0xd5 1M 32k', test_img
)
191 qemu_io('-c', 'write -P0xdc 32M 124k', test_img
)
192 qemu_io('-c', 'write -P0xdc 67043328 64k', test_img
)
194 self
.vm
= iotests
.VM().add_drive(test_img
)
201 os
.remove(target_img
)
205 def test_cancel(self
):
206 self
.assert_no_active_block_jobs()
208 result
= self
.vm
.qmp('transaction', actions
=[{
209 'type': 'drive-backup',
210 'data': { 'device': 'drive0',
211 'target': target_img
,
215 self
.assert_qmp(result
, 'return', {})
217 event
= self
.cancel_and_wait()
218 self
.assert_qmp(event
, 'data/type', 'backup')
220 def test_pause(self
):
221 self
.assert_no_active_block_jobs()
223 self
.vm
.pause_drive('drive0')
224 result
= self
.vm
.qmp('transaction', actions
=[{
225 'type': 'drive-backup',
226 'data': { 'device': 'drive0',
227 'target': target_img
,
231 self
.assert_qmp(result
, 'return', {})
233 result
= self
.vm
.qmp('block-job-pause', device
='drive0')
234 self
.assert_qmp(result
, 'return', {})
236 self
.vm
.resume_drive('drive0')
238 result
= self
.vm
.qmp('query-block-jobs')
239 offset
= self
.dictpath(result
, 'return[0]/offset')
242 result
= self
.vm
.qmp('query-block-jobs')
243 self
.assert_qmp(result
, 'return[0]/offset', offset
)
245 result
= self
.vm
.qmp('block-job-resume', device
='drive0')
246 self
.assert_qmp(result
, 'return', {})
248 self
.wait_until_completed()
251 self
.assertTrue(iotests
.compare_images(test_img
, target_img
),
252 'target image does not match source after backup')
254 def test_medium_not_found(self
):
255 result
= self
.vm
.qmp('transaction', actions
=[{
256 'type': 'drive-backup',
257 'data': { 'device': 'ide1-cd0',
258 'target': target_img
,
262 self
.assert_qmp(result
, 'error/class', 'GenericError')
264 def test_image_not_found(self
):
265 result
= self
.vm
.qmp('transaction', actions
=[{
266 'type': 'drive-backup',
267 'data': { 'device': 'drive0',
269 'target': target_img
,
273 self
.assert_qmp(result
, 'error/class', 'GenericError')
275 def test_device_not_found(self
):
276 result
= self
.vm
.qmp('transaction', actions
=[{
277 'type': 'drive-backup',
278 'data': { 'device': 'nonexistent',
280 'target': target_img
,
284 self
.assert_qmp(result
, 'error/class', 'DeviceNotFound')
286 def test_abort(self
):
287 result
= self
.vm
.qmp('transaction', actions
=[{
288 'type': 'drive-backup',
289 'data': { 'device': 'nonexistent',
291 'target': target_img
,
298 self
.assert_qmp(result
, 'error/class', 'GenericError')
299 self
.assert_no_active_block_jobs()
301 if __name__
== '__main__':
302 iotests
.main(supported_fmts
=['raw', 'qcow2'])