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 result
= self
.vm
.qmp('drive-backup', device
='drive0',
67 target
=target_img
, sync
='full')
68 self
.assert_qmp(result
, 'return', {})
70 result
= self
.vm
.qmp('block-job-pause', device
='drive0')
71 self
.assert_qmp(result
, 'return', {})
74 result
= self
.vm
.qmp('query-block-jobs')
75 offset
= self
.dictpath(result
, 'return[0]/offset')
78 result
= self
.vm
.qmp('query-block-jobs')
79 self
.assert_qmp(result
, 'return[0]/offset', offset
)
81 result
= self
.vm
.qmp('block-job-resume', device
='drive0')
82 self
.assert_qmp(result
, 'return', {})
84 self
.wait_until_completed()
87 self
.assertTrue(iotests
.compare_images(test_img
, target_img
),
88 'target image does not match source after backup')
90 def test_medium_not_found(self
):
91 result
= self
.vm
.qmp('drive-backup', device
='ide1-cd0',
92 target
=target_img
, sync
='full')
93 self
.assert_qmp(result
, 'error/class', 'GenericError')
95 def test_image_not_found(self
):
96 result
= self
.vm
.qmp('drive-backup', device
='drive0',
97 target
=target_img
, sync
='full', mode
='existing')
98 self
.assert_qmp(result
, 'error/class', 'GenericError')
100 def test_invalid_format(self
):
101 result
= self
.vm
.qmp('drive-backup', device
='drive0',
102 target
=target_img
, sync
='full',
103 format
='spaghetti-noodles')
104 self
.assert_qmp(result
, 'error/class', 'GenericError')
106 def test_device_not_found(self
):
107 result
= self
.vm
.qmp('drive-backup', device
='nonexistent',
108 target
=target_img
, sync
='full')
109 self
.assert_qmp(result
, 'error/class', 'DeviceNotFound')
111 class TestSetSpeed(iotests
.QMPTestCase
):
112 image_len
= 80 * 1024 * 1024 # MB
115 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, str(TestSetSpeed
.image_len
))
116 self
.vm
= iotests
.VM().add_drive(test_img
)
122 os
.remove(target_img
)
124 def test_set_speed(self
):
125 self
.assert_no_active_block_jobs()
127 result
= self
.vm
.qmp('drive-backup', device
='drive0',
128 target
=target_img
, sync
='full')
129 self
.assert_qmp(result
, 'return', {})
132 result
= self
.vm
.qmp('query-block-jobs')
133 self
.assert_qmp(result
, 'return[0]/device', 'drive0')
134 self
.assert_qmp(result
, 'return[0]/speed', 0)
136 result
= self
.vm
.qmp('block-job-set-speed', device
='drive0', speed
=8 * 1024 * 1024)
137 self
.assert_qmp(result
, 'return', {})
139 # Ensure the speed we set was accepted
140 result
= self
.vm
.qmp('query-block-jobs')
141 self
.assert_qmp(result
, 'return[0]/device', 'drive0')
142 self
.assert_qmp(result
, 'return[0]/speed', 8 * 1024 * 1024)
144 event
= self
.cancel_and_wait()
145 self
.assert_qmp(event
, 'data/type', 'backup')
147 # Check setting speed in drive-backup works
148 result
= self
.vm
.qmp('drive-backup', device
='drive0',
149 target
=target_img
, sync
='full', speed
=4*1024*1024)
150 self
.assert_qmp(result
, 'return', {})
152 result
= self
.vm
.qmp('query-block-jobs')
153 self
.assert_qmp(result
, 'return[0]/device', 'drive0')
154 self
.assert_qmp(result
, 'return[0]/speed', 4 * 1024 * 1024)
156 event
= self
.cancel_and_wait()
157 self
.assert_qmp(event
, 'data/type', 'backup')
159 def test_set_speed_invalid(self
):
160 self
.assert_no_active_block_jobs()
162 result
= self
.vm
.qmp('drive-backup', device
='drive0',
163 target
=target_img
, sync
='full', speed
=-1)
164 self
.assert_qmp(result
, 'error/class', 'GenericError')
166 self
.assert_no_active_block_jobs()
168 result
= self
.vm
.qmp('drive-backup', device
='drive0',
169 target
=target_img
, sync
='full')
170 self
.assert_qmp(result
, 'return', {})
172 result
= self
.vm
.qmp('block-job-set-speed', device
='drive0', speed
=-1)
173 self
.assert_qmp(result
, 'error/class', 'GenericError')
175 event
= self
.cancel_and_wait()
176 self
.assert_qmp(event
, 'data/type', 'backup')
178 class TestSingleTransaction(iotests
.QMPTestCase
):
179 image_len
= 64 * 1024 * 1024 # MB
182 qemu_img('create', '-f', iotests
.imgfmt
, test_img
, str(TestSingleTransaction
.image_len
))
183 qemu_io('-c', 'write -P0x5d 0 64k', test_img
)
184 qemu_io('-c', 'write -P0xd5 1M 32k', test_img
)
185 qemu_io('-c', 'write -P0xdc 32M 124k', test_img
)
186 qemu_io('-c', 'write -P0xdc 67043328 64k', test_img
)
188 self
.vm
= iotests
.VM().add_drive(test_img
)
195 os
.remove(target_img
)
199 def test_cancel(self
):
200 self
.assert_no_active_block_jobs()
202 result
= self
.vm
.qmp('transaction', actions
=[{
203 'type': 'drive-backup',
204 'data': { 'device': 'drive0',
205 'target': target_img
,
209 self
.assert_qmp(result
, 'return', {})
211 event
= self
.cancel_and_wait()
212 self
.assert_qmp(event
, 'data/type', 'backup')
214 def test_pause(self
):
215 self
.assert_no_active_block_jobs()
217 result
= self
.vm
.qmp('transaction', actions
=[{
218 'type': 'drive-backup',
219 'data': { 'device': 'drive0',
220 'target': target_img
,
224 self
.assert_qmp(result
, 'return', {})
226 result
= self
.vm
.qmp('block-job-pause', device
='drive0')
227 self
.assert_qmp(result
, 'return', {})
230 result
= self
.vm
.qmp('query-block-jobs')
231 offset
= self
.dictpath(result
, 'return[0]/offset')
234 result
= self
.vm
.qmp('query-block-jobs')
235 self
.assert_qmp(result
, 'return[0]/offset', offset
)
237 result
= self
.vm
.qmp('block-job-resume', device
='drive0')
238 self
.assert_qmp(result
, 'return', {})
240 self
.wait_until_completed()
243 self
.assertTrue(iotests
.compare_images(test_img
, target_img
),
244 'target image does not match source after backup')
246 def test_medium_not_found(self
):
247 result
= self
.vm
.qmp('transaction', actions
=[{
248 'type': 'drive-backup',
249 'data': { 'device': 'ide1-cd0',
250 'target': target_img
,
254 self
.assert_qmp(result
, 'error/class', 'GenericError')
256 def test_image_not_found(self
):
257 result
= self
.vm
.qmp('transaction', actions
=[{
258 'type': 'drive-backup',
259 'data': { 'device': 'drive0',
261 'target': target_img
,
265 self
.assert_qmp(result
, 'error/class', 'GenericError')
267 def test_device_not_found(self
):
268 result
= self
.vm
.qmp('transaction', actions
=[{
269 'type': 'drive-backup',
270 'data': { 'device': 'nonexistent',
272 'target': target_img
,
276 self
.assert_qmp(result
, 'error/class', 'DeviceNotFound')
278 def test_abort(self
):
279 result
= self
.vm
.qmp('transaction', actions
=[{
280 'type': 'drive-backup',
281 'data': { 'device': 'nonexistent',
283 'target': target_img
,
290 self
.assert_qmp(result
, 'error/class', 'GenericError')
291 self
.assert_no_active_block_jobs()
293 if __name__
== '__main__':
294 iotests
.main(supported_fmts
=['raw', 'qcow2'])