← Back to branch summary

~ubuntu-branches/ubuntu/utopic/cinder/utopic

« back to all changes in this revision

Viewing changes to cinder/tests/test_image_utils.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, James Page, Adam Gandelman, Chuck Short
  • Date: 2013-09-08 21:09:46 UTC
  • mfrom: (1.1.18)
  • Revision ID: package-import@ubuntu.com-20130908210946-3dbzq1jy5uji4wad
Tags: 1:2013.2~b3-0ubuntu1
https://launchpad.net/bugs/1190791
[ James Page ]
* d/control: Switch ceph-common -> python-ceph inline with upstream
  refactoring of Ceph RBD driver, move to Suggests of python-cinder.
  (LP: #1190791). 

[ Adam Gandelman ]
* debian/patches/avoid_paramiko_vers_depends.patch: Dropped, no longer
  required.
* Add minimum requirement python-greenlet (>= 0.3.2).
* Add minimum requirement python-eventlet (>= 0.12.0).
* Add minimum requirement python-paramiko (>= 1.8).

[ Chuck Short ]
* New upstream release.
* debian/patches/skip-sqlachemy-failures.patch: Skip testfailures
  with sqlalchemy 0.8 until they are fixed upstream.
* debian/control: Add python-babel to build-depends.
* debian/control: Add python-novaclient to build-depends.

Show diffs side-by-side

added added

removed removed

Lines of Context:
cinder/tests/test_image_utils.py
18
18
 
19
19
import contextlib
20
20
import mox
 
21
import tempfile
21
22
import textwrap
22
23
 
 
24
from cinder import context
 
25
from cinder import exception
23
26
from cinder.image import image_utils
24
27
from cinder import test
25
28
from cinder import utils
26
29
 
27
30
 
 
31
class FakeImageService:
 
32
    def __init__(self):
 
33
        self._imagedata = {}
 
34
 
 
35
    def download(self, context, image_id, data):
 
36
        self.show(context, image_id)
 
37
        data.write(self._imagedata.get(image_id, ''))
 
38
 
 
39
    def show(self, context, image_id):
 
40
        return {'size': 2 * 1024 * 1024 * 1024,
 
41
                'disk_format': 'qcow2',
 
42
                'container_format': 'bare'}
 
43
 
 
44
    def update(self, context, image_id, metadata, path):
 
45
        pass
 
46
 
 
47
 
28
48
class TestUtils(test.TestCase):
 
49
    TEST_IMAGE_ID = 321
 
50
    TEST_DEV_PATH = "/dev/ether/fake_dev"
 
51
 
29
52
    def setUp(self):
30
53
        super(TestUtils, self).setUp()
31
54
        self._mox = mox.Mox()
47
70
 
48
71
        mox.VerifyAll()
49
72
 
 
73
    def test_convert_image(self):
 
74
        mox = self._mox
 
75
        mox.StubOutWithMock(utils, 'execute')
 
76
 
 
77
        TEST_OUT_FORMAT = 'vmdk'
 
78
        TEST_SOURCE = 'img/qemu.img'
 
79
        TEST_DEST = '/img/vmware.vmdk'
 
80
 
 
81
        utils.execute('qemu-img', 'convert', '-O', TEST_OUT_FORMAT,
 
82
                      TEST_SOURCE, TEST_DEST, run_as_root=True)
 
83
 
 
84
        mox.ReplayAll()
 
85
 
 
86
        image_utils.convert_image(TEST_SOURCE, TEST_DEST, TEST_OUT_FORMAT)
 
87
 
 
88
        mox.VerifyAll()
 
89
 
 
90
    def test_qemu_img_info(self):
 
91
        TEST_PATH = "img/qemu.qcow2"
 
92
        TEST_RETURN = "image: qemu.qcow2\n"\
 
93
                      "backing_file: qemu.qcow2 (actual path: qemu.qcow2)\n"\
 
94
                      "file_format: qcow2\n"\
 
95
                      "virtual_size: 50M (52428800 bytes)\n"\
 
96
                      "cluster_size: 65536\n"\
 
97
                      "disk_size: 196K (200704 bytes)\n"\
 
98
                      "Snapshot list:\n"\
 
99
                      "ID TAG  VM SIZE DATE VM CLOCK\n"\
 
100
                      "1  snap1 1.7G 2011-10-04 19:04:00 32:06:34.974"
 
101
        TEST_STR = "image: qemu.qcow2\n"\
 
102
                   "file_format: qcow2\n"\
 
103
                   "virtual_size: 52428800\n"\
 
104
                   "disk_size: 200704\n"\
 
105
                   "cluster_size: 65536\n"\
 
106
                   "backing_file: qemu.qcow2\n"\
 
107
                   "snapshots: [{'date': '2011-10-04', "\
 
108
                   "'vm_clock': '19:04:00 32:06:34.974', "\
 
109
                   "'vm_size': '1.7G', 'tag': 'snap1', 'id': '1'}]"
 
110
 
 
111
        mox = self._mox
 
112
        mox.StubOutWithMock(utils, 'execute')
 
113
 
 
114
        utils.execute(
 
115
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
116
            TEST_PATH, run_as_root=True).AndReturn(
 
117
                (TEST_RETURN, 'ignored')
 
118
            )
 
119
 
 
120
        mox.ReplayAll()
 
121
 
 
122
        inf = image_utils.qemu_img_info(TEST_PATH)
 
123
 
 
124
        self.assertEquals(inf.image, 'qemu.qcow2')
 
125
        self.assertEquals(inf.backing_file, 'qemu.qcow2')
 
126
        self.assertEquals(inf.file_format, 'qcow2')
 
127
        self.assertEquals(inf.virtual_size, 52428800)
 
128
        self.assertEquals(inf.cluster_size, 65536)
 
129
        self.assertEquals(inf.disk_size, 200704)
 
130
 
 
131
        self.assertEquals(inf.snapshots[0]['id'], '1')
 
132
        self.assertEquals(inf.snapshots[0]['tag'], 'snap1')
 
133
        self.assertEquals(inf.snapshots[0]['vm_size'], '1.7G')
 
134
        self.assertEquals(inf.snapshots[0]['date'], '2011-10-04')
 
135
        self.assertEquals(inf.snapshots[0]['vm_clock'],
 
136
                          '19:04:00 32:06:34.974')
 
137
 
 
138
        self.assertEquals(str(inf), TEST_STR)
 
139
 
 
140
    def test_fetch_to_raw(self):
 
141
        TEST_RET = "image: qemu.qcow2\n"\
 
142
                   "file_format: qcow2 \n"\
 
143
                   "virtual_size: 50M (52428800 bytes)\n"\
 
144
                   "cluster_size: 65536\n"\
 
145
                   "disk_size: 196K (200704 bytes)"
 
146
        TEST_RETURN_RAW = "image: qemu.raw\n"\
 
147
                          "file_format: raw\n"\
 
148
                          "virtual_size: 50M (52428800 bytes)\n"\
 
149
                          "cluster_size: 65536\n"\
 
150
                          "disk_size: 196K (200704 bytes)\n"\
 
151
 
 
152
        fake_image_service = FakeImageService()
 
153
        mox = self._mox
 
154
 
 
155
        mox.StubOutWithMock(image_utils, 'create_temporary_file')
 
156
        mox.StubOutWithMock(utils, 'execute')
 
157
        mox.StubOutWithMock(image_utils, 'fetch')
 
158
 
 
159
        image_utils.create_temporary_file().AndReturn(self.TEST_DEV_PATH)
 
160
        image_utils.fetch(context, fake_image_service,
 
161
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None)
 
162
 
 
163
        utils.execute(
 
164
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
165
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
166
                (TEST_RET, 'ignored')
 
167
            )
 
168
 
 
169
        utils.execute('qemu-img', 'convert', '-O', 'raw',
 
170
                      self.TEST_DEV_PATH, self.TEST_DEV_PATH, run_as_root=True)
 
171
 
 
172
        utils.execute(
 
173
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
174
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
175
                (TEST_RETURN_RAW, 'ignored')
 
176
            )
 
177
 
 
178
        mox.ReplayAll()
 
179
 
 
180
        image_utils.fetch_to_raw(context, fake_image_service,
 
181
                                 self.TEST_IMAGE_ID, self.TEST_DEV_PATH)
 
182
        mox.VerifyAll()
 
183
 
 
184
    def test_fetch_to_raw_on_error_parsing_failed(self):
 
185
        TEST_RET = "image: qemu.qcow2\n"\
 
186
                   "virtual_size: 50M (52428800 bytes)\n"\
 
187
                   "cluster_size: 65536\n"\
 
188
                   "disk_size: 196K (200704 bytes)"
 
189
 
 
190
        fake_image_service = FakeImageService()
 
191
        mox = self._mox
 
192
 
 
193
        mox.StubOutWithMock(image_utils, 'create_temporary_file')
 
194
        mox.StubOutWithMock(utils, 'execute')
 
195
        mox.StubOutWithMock(image_utils, 'fetch')
 
196
 
 
197
        image_utils.create_temporary_file().AndReturn(self.TEST_DEV_PATH)
 
198
        image_utils.fetch(context, fake_image_service,
 
199
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None)
 
200
 
 
201
        utils.execute(
 
202
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
203
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
204
                (TEST_RET, 'ignored')
 
205
            )
 
206
 
 
207
        mox.ReplayAll()
 
208
 
 
209
        self.assertRaises(exception.ImageUnacceptable,
 
210
                          image_utils.fetch_to_raw, context,
 
211
                          fake_image_service, self.TEST_IMAGE_ID,
 
212
                          self.TEST_DEV_PATH)
 
213
 
 
214
    def test_fetch_to_raw_on_error_backing_file(self):
 
215
        TEST_RET = "image: qemu.qcow2\n"\
 
216
                   "backing_file: qemu.qcow2 (actual path: qemu.qcow2)\n"\
 
217
                   "file_format: qcow2 \n"\
 
218
                   "virtual_size: 50M (52428800 bytes)\n"\
 
219
                   "cluster_size: 65536\n"\
 
220
                   "disk_size: 196K (200704 bytes)"
 
221
 
 
222
        fake_image_service = FakeImageService()
 
223
        mox = self._mox
 
224
 
 
225
        mox.StubOutWithMock(image_utils, 'create_temporary_file')
 
226
        mox.StubOutWithMock(utils, 'execute')
 
227
        mox.StubOutWithMock(image_utils, 'fetch')
 
228
 
 
229
        image_utils.create_temporary_file().AndReturn(self.TEST_DEV_PATH)
 
230
        image_utils.fetch(context, fake_image_service,
 
231
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None)
 
232
 
 
233
        utils.execute(
 
234
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
235
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
236
                (TEST_RET, 'ignored')
 
237
            )
 
238
 
 
239
        mox.ReplayAll()
 
240
        self.assertRaises(exception.ImageUnacceptable,
 
241
                          image_utils.fetch_to_raw,
 
242
                          context, fake_image_service,
 
243
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH)
 
244
 
 
245
    def test_fetch_to_raw_on_error_not_convert_to_raw(self):
 
246
        TEST_RET = "image: qemu.qcow2\n"\
 
247
                   "file_format: qcow2 \n"\
 
248
                   "virtual_size: 50M (52428800 bytes)\n"\
 
249
                   "cluster_size: 65536\n"\
 
250
                   "disk_size: 196K (200704 bytes)"
 
251
 
 
252
        fake_image_service = FakeImageService()
 
253
        mox = self._mox
 
254
 
 
255
        mox.StubOutWithMock(image_utils, 'create_temporary_file')
 
256
        mox.StubOutWithMock(utils, 'execute')
 
257
        mox.StubOutWithMock(image_utils, 'fetch')
 
258
 
 
259
        image_utils.create_temporary_file().AndReturn(self.TEST_DEV_PATH)
 
260
        image_utils.fetch(context, fake_image_service,
 
261
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None)
 
262
 
 
263
        utils.execute(
 
264
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
265
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
266
                (TEST_RET, 'ignored')
 
267
            )
 
268
 
 
269
        utils.execute('qemu-img', 'convert', '-O', 'raw',
 
270
                      self.TEST_DEV_PATH, self.TEST_DEV_PATH, run_as_root=True)
 
271
 
 
272
        utils.execute(
 
273
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
274
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
275
                (TEST_RET, 'ignored')
 
276
            )
 
277
 
 
278
        mox.ReplayAll()
 
279
 
 
280
        self.assertRaises(exception.ImageUnacceptable,
 
281
                          image_utils.fetch_to_raw,
 
282
                          context, fake_image_service,
 
283
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH)
 
284
 
 
285
    def test_fetch_verify_image_with_backing_file(self):
 
286
        TEST_RETURN = "image: qemu.qcow2\n"\
 
287
                      "backing_file: qemu.qcow2 (actual path: qemu.qcow2)\n"\
 
288
                      "file_format: qcow2\n"\
 
289
                      "virtual_size: 50M (52428800 bytes)\n"\
 
290
                      "cluster_size: 65536\n"\
 
291
                      "disk_size: 196K (200704 bytes)\n"\
 
292
                      "Snapshot list:\n"\
 
293
                      "ID TAG  VM SIZE DATE VM CLOCK\n"\
 
294
                      "1  snap1 1.7G 2011-10-04 19:04:00 32:06:34.974"
 
295
 
 
296
        fake_image_service = FakeImageService()
 
297
        mox = self._mox
 
298
        mox.StubOutWithMock(image_utils, 'fetch')
 
299
        mox.StubOutWithMock(utils, 'execute')
 
300
        image_utils.fetch(context, fake_image_service,
 
301
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None)
 
302
 
 
303
        utils.execute(
 
304
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
305
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
306
                (TEST_RETURN, 'ignored')
 
307
            )
 
308
 
 
309
        mox.ReplayAll()
 
310
 
 
311
        self.assertRaises(exception.ImageUnacceptable,
 
312
                          image_utils.fetch_verify_image,
 
313
                          context, fake_image_service,
 
314
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH)
 
315
 
 
316
    def test_fetch_verify_image_without_file_format(self):
 
317
        TEST_RETURN = "image: qemu.qcow2\n"\
 
318
                      "virtual_size: 50M (52428800 bytes)\n"\
 
319
                      "cluster_size: 65536\n"\
 
320
                      "disk_size: 196K (200704 bytes)\n"\
 
321
                      "Snapshot list:\n"\
 
322
                      "ID TAG  VM SIZE DATE VM CLOCK\n"\
 
323
                      "1  snap1 1.7G 2011-10-04 19:04:00 32:06:34.974"
 
324
 
 
325
        fake_image_service = FakeImageService()
 
326
        mox = self._mox
 
327
        mox.StubOutWithMock(image_utils, 'fetch')
 
328
        mox.StubOutWithMock(utils, 'execute')
 
329
        image_utils.fetch(context, fake_image_service,
 
330
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH, None, None)
 
331
 
 
332
        utils.execute(
 
333
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
334
            self.TEST_DEV_PATH, run_as_root=True).AndReturn(
 
335
                (TEST_RETURN, 'ignored')
 
336
            )
 
337
 
 
338
        mox.ReplayAll()
 
339
 
 
340
        self.assertRaises(exception.ImageUnacceptable,
 
341
                          image_utils.fetch_verify_image,
 
342
                          context, fake_image_service,
 
343
                          self.TEST_IMAGE_ID, self.TEST_DEV_PATH)
 
344
 
 
345
    def test_upload_volume(self):
 
346
        image_meta = {'id': 1, 'disk_format': 'qcow2'}
 
347
        TEST_RET = "image: qemu.qcow2\n"\
 
348
                   "file_format: qcow2 \n"\
 
349
                   "virtual_size: 50M (52428800 bytes)\n"\
 
350
                   "cluster_size: 65536\n"\
 
351
                   "disk_size: 196K (200704 bytes)"
 
352
 
 
353
        m = self._mox
 
354
        m.StubOutWithMock(utils, 'execute')
 
355
 
 
356
        utils.execute('qemu-img', 'convert', '-O', 'qcow2',
 
357
                      mox.IgnoreArg(), mox.IgnoreArg(), run_as_root=True)
 
358
        utils.execute(
 
359
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
360
            mox.IgnoreArg(), run_as_root=True).AndReturn(
 
361
                (TEST_RET, 'ignored')
 
362
            )
 
363
 
 
364
        m.ReplayAll()
 
365
 
 
366
        image_utils.upload_volume(context, FakeImageService(),
 
367
                                  image_meta, '/dev/loop1')
 
368
        m.VerifyAll()
 
369
 
 
370
    def test_upload_volume_with_raw_image(self):
 
371
        image_meta = {'id': 1, 'disk_format': 'raw'}
 
372
        mox = self._mox
 
373
 
 
374
        mox.StubOutWithMock(image_utils, 'convert_image')
 
375
 
 
376
        mox.ReplayAll()
 
377
 
 
378
        with tempfile.NamedTemporaryFile() as f:
 
379
            image_utils.upload_volume(context, FakeImageService(),
 
380
                                      image_meta, f.name)
 
381
        mox.VerifyAll()
 
382
 
 
383
    def test_upload_volume_on_error(self):
 
384
        image_meta = {'id': 1, 'disk_format': 'qcow2'}
 
385
        TEST_RET = "image: qemu.vhd\n"\
 
386
                   "file_format: vhd \n"\
 
387
                   "virtual_size: 50M (52428800 bytes)\n"\
 
388
                   "cluster_size: 65536\n"\
 
389
                   "disk_size: 196K (200704 bytes)"
 
390
 
 
391
        m = self._mox
 
392
        m.StubOutWithMock(utils, 'execute')
 
393
 
 
394
        utils.execute('qemu-img', 'convert', '-O', 'qcow2',
 
395
                      mox.IgnoreArg(), mox.IgnoreArg(), run_as_root=True)
 
396
        utils.execute(
 
397
            'env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info',
 
398
            mox.IgnoreArg(), run_as_root=True).AndReturn(
 
399
                (TEST_RET, 'ignored')
 
400
            )
 
401
 
 
402
        m.ReplayAll()
 
403
 
 
404
        self.assertRaises(exception.ImageUnacceptable,
 
405
                          image_utils.upload_volume,
 
406
                          context, FakeImageService(),
 
407
                          image_meta, '/dev/loop1')
 
408
        m.VerifyAll()
 
409
 
50
410
 
51
411
class TestExtractTo(test.TestCase):
52
412
    def test_extract_to_calls_tar(self):
184
544
 
185
545
        def sut():
186
546
            with image_utils.temporary_file():
187
 
                raise Exception()
 
547
                raise test.TestingException()
188
548
 
189
 
        self.assertRaises(Exception, sut)
 
549
        self.assertRaises(test.TestingException, sut)
190
550
 
191
551
 
192
552
class TestCoalesceChain(test.TestCase):