1# Copyright 2010 United States Government as represented by the 2# Administrator of the National Aeronautics and Space Administration. 3# All Rights Reserved. 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); you may 6# not use this file except in compliance with the License. You may obtain 7# a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 14# License for the specific language governing permissions and limitations 15# under the License. 16"""Tests for volume and images.""" 17 18import datetime 19import mock 20import os 21import tempfile 22 23from oslo_utils import imageutils 24from oslo_utils import units 25 26from cinder import db 27from cinder import exception 28from cinder.message import message_field 29from cinder import objects 30from cinder.objects import fields 31from cinder import quota 32from cinder.tests import fake_driver 33from cinder.tests.unit import fake_constants as fake 34from cinder.tests.unit.image import fake as fake_image 35from cinder.tests.unit import utils as tests_utils 36from cinder.tests.unit import volume as base 37import cinder.volume 38from cinder.volume import manager as vol_manager 39 40QUOTAS = quota.QUOTAS 41NON_EXISTENT_IMAGE_ID = '003f540f-ec6b-4293-a3f9-7c68646b0f5c' 42 43 44class FakeImageService(object): 45 def __init__(self, db_driver=None, image_service=None): 46 pass 47 48 def show(self, context, image_id): 49 return {'size': 2 * units.Gi, 50 'disk_format': 'raw', 51 'container_format': 'bare', 52 'status': 'active'} 53 54 55class CopyVolumeToImageTestCase(base.BaseVolumeTestCase): 56 def fake_local_path(self, volume): 57 return self.dst_path 58 59 def setUp(self): 60 super(CopyVolumeToImageTestCase, self).setUp() 61 self.dst_fd, self.dst_path = tempfile.mkstemp() 62 self.addCleanup(os.unlink, self.dst_path) 63 os.close(self.dst_fd) 64 self.mock_object(self.volume.driver, 'local_path', 65 self.fake_local_path) 66 self.mock_cache = mock.MagicMock() 67 self.image_id = '70a599e0-31e7-49b7-b260-868f441e862b' 68 self.image_meta = { 69 'id': self.image_id, 70 'container_format': 'bare', 71 'disk_format': 'raw' 72 } 73 self.volume_id = fake.VOLUME_ID 74 self.addCleanup(db.volume_destroy, self.context, self.volume_id) 75 76 self.volume_attrs = { 77 'id': self.volume_id, 78 'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1), 79 'display_description': 'Test Desc', 80 'size': 20, 81 'status': 'uploading', 82 'host': 'dummy' 83 } 84 85 def test_copy_volume_to_image_status_available(self): 86 # creating volume testdata 87 self.volume_attrs['instance_uuid'] = None 88 db.volume_create(self.context, self.volume_attrs) 89 90 # start test 91 self.volume.copy_volume_to_image(self.context, 92 self.volume_id, 93 self.image_meta) 94 95 volume = db.volume_get(self.context, self.volume_id) 96 self.assertEqual('available', volume['status']) 97 98 def test_copy_volume_to_image_over_image_quota(self): 99 # creating volume testdata 100 self.volume_attrs['instance_uuid'] = None 101 volume = db.volume_create(self.context, self.volume_attrs) 102 103 with mock.patch.object(self.volume.driver, 104 'copy_volume_to_image') as driver_copy_mock: 105 driver_copy_mock.side_effect = exception.ImageLimitExceeded 106 107 # test with image not in queued state 108 self.assertRaises(exception.ImageLimitExceeded, 109 self.volume.copy_volume_to_image, 110 self.context, 111 self.volume_id, 112 self.image_meta) 113 # Assert a user message was created 114 self.volume.message_api.create.assert_called_once_with( 115 self.context, 116 message_field.Action.COPY_VOLUME_TO_IMAGE, 117 resource_uuid=volume['id'], 118 exception=mock.ANY, 119 detail=message_field.Detail.FAILED_TO_UPLOAD_VOLUME) 120 121 def test_copy_volume_to_image_instance_deleted(self): 122 # During uploading volume to image if instance is deleted, 123 # volume should be in available status. 124 self.image_meta['id'] = 'a440c04b-79fa-479c-bed1-0b816eaec379' 125 # Creating volume testdata 126 self.volume_attrs['instance_uuid'] = 'b21f957d-a72f-4b93-b5a5-' \ 127 '45b1161abb02' 128 db.volume_create(self.context, self.volume_attrs) 129 130 method = 'volume_update_status_based_on_attachment' 131 with mock.patch.object(db, method, 132 wraps=getattr(db, method)) as mock_update: 133 # Start test 134 self.volume.copy_volume_to_image(self.context, 135 self.volume_id, 136 self.image_meta) 137 # Check 'volume_update_status_after_copy_volume_to_image' 138 # is called 1 time 139 self.assertEqual(1, mock_update.call_count) 140 141 # Check volume status has changed to available because 142 # instance is deleted 143 volume = db.volume_get(self.context, self.volume_id) 144 self.assertEqual('available', volume['status']) 145 146 def test_copy_volume_to_image_status_use(self): 147 self.image_meta['id'] = 'a440c04b-79fa-479c-bed1-0b816eaec379' 148 # creating volume testdata 149 db.volume_create(self.context, self.volume_attrs) 150 151 # start test 152 self.volume.copy_volume_to_image(self.context, 153 self.volume_id, 154 self.image_meta) 155 156 volume = db.volume_get(self.context, self.volume_id) 157 self.assertEqual('available', volume['status']) 158 159 def test_copy_volume_to_image_exception(self): 160 self.image_meta['id'] = NON_EXISTENT_IMAGE_ID 161 # creating volume testdata 162 self.volume_attrs['status'] = 'in-use' 163 db.volume_create(self.context, self.volume_attrs) 164 165 # start test 166 self.assertRaises(exception.ImageNotFound, 167 self.volume.copy_volume_to_image, 168 self.context, 169 self.volume_id, 170 self.image_meta) 171 172 volume = db.volume_get(self.context, self.volume_id) 173 self.assertEqual('available', volume['status']) 174 175 def test_copy_volume_to_image_driver_not_initialized(self): 176 # creating volume testdata 177 db.volume_create(self.context, self.volume_attrs) 178 179 # set initialized to False 180 self.volume.driver._initialized = False 181 182 # start test 183 self.assertRaises(exception.DriverNotInitialized, 184 self.volume.copy_volume_to_image, 185 self.context, 186 self.volume_id, 187 self.image_meta) 188 189 volume = db.volume_get(self.context, self.volume_id) 190 self.assertEqual('available', volume.status) 191 192 def test_copy_volume_to_image_driver_exception(self): 193 self.image_meta['id'] = self.image_id 194 195 image_service = fake_image.FakeImageService() 196 # create new image in queued state 197 queued_image_id = 'd5133f15-f753-41bd-920a-06b8c49275d9' 198 queued_image_meta = image_service.show(self.context, self.image_id) 199 queued_image_meta['id'] = queued_image_id 200 queued_image_meta['status'] = 'queued' 201 image_service.create(self.context, queued_image_meta) 202 203 # create new image in saving state 204 saving_image_id = '5c6eec33-bab4-4e7d-b2c9-88e2d0a5f6f2' 205 saving_image_meta = image_service.show(self.context, self.image_id) 206 saving_image_meta['id'] = saving_image_id 207 saving_image_meta['status'] = 'saving' 208 image_service.create(self.context, saving_image_meta) 209 210 # create volume 211 self.volume_attrs['status'] = 'available' 212 self.volume_attrs['instance_uuid'] = None 213 db.volume_create(self.context, self.volume_attrs) 214 215 with mock.patch.object(self.volume.driver, 216 'copy_volume_to_image') as driver_copy_mock: 217 driver_copy_mock.side_effect = exception.VolumeDriverException( 218 "Error") 219 220 # test with image not in queued state 221 self.assertRaises(exception.VolumeDriverException, 222 self.volume.copy_volume_to_image, 223 self.context, 224 self.volume_id, 225 self.image_meta) 226 # Make sure we are passing an OVO instance and not an ORM instance 227 # to the driver 228 self.assertIsInstance(driver_copy_mock.call_args[0][1], 229 objects.Volume) 230 volume = db.volume_get(self.context, self.volume_id) 231 self.assertEqual('available', volume['status']) 232 # image shouldn't be deleted if it is not in queued state 233 image_service.show(self.context, self.image_id) 234 235 # test with image in queued state 236 self.assertRaises(exception.VolumeDriverException, 237 self.volume.copy_volume_to_image, 238 self.context, 239 self.volume_id, 240 queued_image_meta) 241 volume = db.volume_get(self.context, self.volume_id) 242 self.assertEqual('available', volume['status']) 243 # queued image should be deleted 244 self.assertRaises(exception.ImageNotFound, 245 image_service.show, 246 self.context, 247 queued_image_id) 248 249 # test with image in saving state 250 self.assertRaises(exception.VolumeDriverException, 251 self.volume.copy_volume_to_image, 252 self.context, 253 self.volume_id, 254 saving_image_meta) 255 volume = db.volume_get(self.context, self.volume_id) 256 self.assertEqual('available', volume['status']) 257 # image in saving state should be deleted 258 self.assertRaises(exception.ImageNotFound, 259 image_service.show, 260 self.context, 261 saving_image_id) 262 263 @mock.patch.object(QUOTAS, 'reserve') 264 @mock.patch.object(QUOTAS, 'commit') 265 @mock.patch.object(vol_manager.VolumeManager, 'create_volume') 266 @mock.patch.object(fake_driver.FakeLoggingVolumeDriver, 267 'copy_volume_to_image') 268 def _test_copy_volume_to_image_with_image_volume( 269 self, mock_copy, mock_create, mock_quota_commit, 270 mock_quota_reserve): 271 self.volume.driver.configuration.image_upload_use_cinder_backend = True 272 self.addCleanup(fake_image.FakeImageService_reset) 273 image_service = fake_image.FakeImageService() 274 275 def add_location_wrapper(ctx, id, uri, metadata): 276 try: 277 volume = db.volume_get(ctx, id) 278 self.assertEqual(ctx.project_id, 279 volume['metadata']['image_owner']) 280 except exception.VolumeNotFound: 281 pass 282 return image_service.add_location_orig(ctx, id, uri, metadata) 283 284 image_service.add_location_orig = image_service.add_location 285 image_service.add_location = add_location_wrapper 286 287 image_id = '5c6eec33-bab4-4e7d-b2c9-88e2d0a5f6f2' 288 self.image_meta['id'] = image_id 289 self.image_meta['status'] = 'queued' 290 image_service.create(self.context, self.image_meta) 291 292 # creating volume testdata 293 self.volume_attrs['instance_uuid'] = None 294 db.volume_create(self.context, self.volume_attrs) 295 296 def fake_create(context, volume, **kwargs): 297 db.volume_update(context, volume.id, {'status': 'available'}) 298 299 mock_create.side_effect = fake_create 300 301 # start test 302 self.volume.copy_volume_to_image(self.context, 303 self.volume_id, 304 self.image_meta) 305 306 volume = db.volume_get(self.context, self.volume_id) 307 self.assertEqual('available', volume['status']) 308 309 # return create image 310 image = image_service.show(self.context, image_id) 311 image_service.delete(self.context, image_id) 312 return image 313 314 def test_copy_volume_to_image_with_image_volume(self): 315 image = self._test_copy_volume_to_image_with_image_volume() 316 self.assertTrue(image['locations'][0]['url'].startswith('cinder://')) 317 318 def test_copy_volume_to_image_with_image_volume_qcow2(self): 319 self.image_meta['disk_format'] = 'qcow2' 320 image = self._test_copy_volume_to_image_with_image_volume() 321 self.assertNotIn('locations', image) 322 323 @mock.patch.object(vol_manager.VolumeManager, 'delete_volume') 324 @mock.patch.object(fake_image._FakeImageService, 'add_location', 325 side_effect=exception.Invalid) 326 def test_copy_volume_to_image_with_image_volume_failure( 327 self, mock_add_location, mock_delete): 328 image = self._test_copy_volume_to_image_with_image_volume() 329 self.assertNotIn('locations', image) 330 self.assertTrue(mock_delete.called) 331 332 @mock.patch('cinder.volume.manager.' 333 'VolumeManager._clone_image_volume') 334 @mock.patch('cinder.volume.manager.' 335 'VolumeManager._create_image_cache_volume_entry') 336 def test_create_image_cache_volume_entry(self, 337 mock_cache_entry, 338 mock_clone_image_volume): 339 image_id = self.image_id 340 image_meta = self.image_meta 341 342 self.mock_cache.get_entry.return_value = mock_cache_entry 343 344 if mock_cache_entry: 345 # Entry is in cache, so basically don't do anything. 346 # Make sure we didn't try and create a cache entry 347 self.assertFalse(self.mock_cache.ensure_space.called) 348 self.assertFalse(self.mock_cache.create_cache_entry.called) 349 else: 350 result = self.volume._create_image_cache_volume_entry( 351 self.context, mock_clone_image_volume, image_id, image_meta) 352 self.assertNotEqual(False, result) 353 cache_entry = self.image_volume_cache.get_entry( 354 self.context, mock_clone_image_volume, image_id, image_meta) 355 self.assertIsNotNone(cache_entry) 356 357 358class ImageVolumeCacheTestCase(base.BaseVolumeTestCase): 359 360 def setUp(self): 361 super(ImageVolumeCacheTestCase, self).setUp() 362 self.volume.driver.set_initialized() 363 364 @mock.patch('oslo_utils.importutils.import_object') 365 def test_cache_configs(self, mock_import_object): 366 opts = { 367 'image_volume_cache_enabled': True, 368 'image_volume_cache_max_size_gb': 100, 369 'image_volume_cache_max_count': 20 370 } 371 372 def conf_get(option): 373 if option in opts: 374 return opts[option] 375 else: 376 return None 377 378 mock_driver = mock.Mock() 379 mock_driver.configuration.safe_get.side_effect = conf_get 380 mock_driver.configuration.extra_capabilities = 'null' 381 382 def import_obj(*args, **kwargs): 383 return mock_driver 384 385 mock_import_object.side_effect = import_obj 386 387 manager = vol_manager.VolumeManager(volume_driver=mock_driver) 388 self.assertIsNotNone(manager) 389 self.assertIsNotNone(manager.image_volume_cache) 390 self.assertEqual(100, manager.image_volume_cache.max_cache_size_gb) 391 self.assertEqual(20, manager.image_volume_cache.max_cache_size_count) 392 393 def test_delete_image_volume(self): 394 volume_params = { 395 'status': 'creating', 396 'host': 'some_host', 397 'cluster_name': 'some_cluster', 398 'size': 1 399 } 400 volume_api = cinder.volume.api.API() 401 volume = tests_utils.create_volume(self.context, **volume_params) 402 volume.status = 'available' 403 volume.save() 404 image_id = '70a599e0-31e7-49b7-b260-868f441e862b' 405 db.image_volume_cache_create(self.context, 406 volume['host'], 407 volume_params['cluster_name'], 408 image_id, 409 datetime.datetime.utcnow(), 410 volume['id'], 411 volume['size']) 412 volume_api.delete(self.context, volume) 413 entry = db.image_volume_cache_get_by_volume_id(self.context, 414 volume['id']) 415 self.assertIsNone(entry) 416 417 def test_delete_volume_with_keymanager_exception(self): 418 volume_params = { 419 'host': 'some_host', 420 'size': 1 421 } 422 volume_api = cinder.volume.api.API() 423 volume = tests_utils.create_volume(self.context, **volume_params) 424 425 with mock.patch.object( 426 volume_api.key_manager, 'delete') as key_del_mock: 427 key_del_mock.side_effect = Exception("Key not found") 428 volume_api.delete(self.context, volume) 429 430 431class ImageVolumeTestCases(base.BaseVolumeTestCase): 432 433 @mock.patch('cinder.volume.drivers.lvm.LVMVolumeDriver.' 434 'create_cloned_volume') 435 @mock.patch('cinder.quota.QUOTAS.rollback') 436 @mock.patch('cinder.quota.QUOTAS.commit') 437 @mock.patch('cinder.quota.QUOTAS.reserve', return_value=["RESERVATION"]) 438 def test_clone_image_volume(self, mock_reserve, mock_commit, 439 mock_rollback, mock_cloned_volume): 440 vol = tests_utils.create_volume(self.context, 441 **self.volume_params) 442 # unnecessary attributes should be removed from image volume 443 vol.consistencygroup = None 444 result = self.volume._clone_image_volume(self.context, vol, 445 {'id': fake.VOLUME_ID}) 446 447 self.assertNotEqual(False, result) 448 mock_reserve.assert_called_once_with(self.context, volumes=1, 449 gigabytes=vol.size) 450 mock_commit.assert_called_once_with(self.context, ["RESERVATION"], 451 project_id=vol.project_id) 452 453 @mock.patch('cinder.quota.QUOTAS.rollback') 454 @mock.patch('cinder.quota.QUOTAS.commit') 455 @mock.patch('cinder.quota.QUOTAS.reserve', return_value=["RESERVATION"]) 456 def test_clone_image_volume_creation_failure(self, mock_reserve, 457 mock_commit, mock_rollback): 458 vol = tests_utils.create_volume(self.context, **self.volume_params) 459 with mock.patch.object(objects, 'Volume', side_effect=ValueError): 460 self.assertIsNone(self.volume._clone_image_volume( 461 self.context, vol, {'id': fake.VOLUME_ID})) 462 463 mock_reserve.assert_called_once_with(self.context, volumes=1, 464 gigabytes=vol.size) 465 mock_rollback.assert_called_once_with(self.context, ["RESERVATION"]) 466 467 @mock.patch('cinder.image.image_utils.qemu_img_info') 468 def test_create_volume_from_image_cloned_status_available( 469 self, mock_qemu_info): 470 """Test create volume from image via cloning. 471 472 Verify that after cloning image to volume, it is in available 473 state and is bootable. 474 """ 475 image_info = imageutils.QemuImgInfo() 476 image_info.virtual_size = '1073741824' 477 mock_qemu_info.return_value = image_info 478 479 volume = self._create_volume_from_image() 480 self.assertEqual('available', volume['status']) 481 self.assertTrue(volume['bootable']) 482 self.volume.delete_volume(self.context, volume) 483 484 @mock.patch('cinder.image.image_utils.qemu_img_info') 485 def test_create_volume_from_image_not_cloned_status_available( 486 self, mock_qemu_info): 487 """Test create volume from image via full copy. 488 489 Verify that after copying image to volume, it is in available 490 state and is bootable. 491 """ 492 image_info = imageutils.QemuImgInfo() 493 image_info.virtual_size = '1073741824' 494 mock_qemu_info.return_value = image_info 495 496 volume = self._create_volume_from_image(fakeout_clone_image=True) 497 self.assertEqual('available', volume['status']) 498 self.assertTrue(volume['bootable']) 499 self.volume.delete_volume(self.context, volume) 500 501 def test_create_volume_from_image_exception(self): 502 """Test create volume from a non-existing image. 503 504 Verify that create volume from a non-existing image, the volume 505 status is 'error' and is not bootable. 506 """ 507 dst_fd, dst_path = tempfile.mkstemp() 508 os.close(dst_fd) 509 510 self.mock_object(self.volume.driver, 'local_path', lambda x: dst_path) 511 512 # creating volume testdata 513 kwargs = {'display_description': 'Test Desc', 514 'size': 20, 515 'availability_zone': 'fake_availability_zone', 516 'status': 'creating', 517 'attach_status': fields.VolumeAttachStatus.DETACHED, 518 'host': 'dummy'} 519 volume = objects.Volume(context=self.context, **kwargs) 520 volume.create() 521 522 self.assertRaises(exception.ImageNotFound, 523 self.volume.create_volume, 524 self.context, 525 volume, 526 {'image_id': NON_EXISTENT_IMAGE_ID}) 527 volume = objects.Volume.get_by_id(self.context, volume.id) 528 self.assertEqual("error", volume['status']) 529 self.assertFalse(volume['bootable']) 530 # cleanup 531 volume.destroy() 532 os.unlink(dst_path) 533 534 @mock.patch('cinder.image.image_utils.qemu_img_info') 535 def test_create_volume_from_image_copy_exception_rescheduling( 536 self, mock_qemu_info): 537 """Test create volume with ImageCopyFailure 538 539 This exception should not trigger rescheduling and allocated_capacity 540 should be incremented so we're having assert for that here. 541 """ 542 image_info = imageutils.QemuImgInfo() 543 image_info.virtual_size = '1073741824' 544 mock_qemu_info.return_value = image_info 545 546 def fake_copy_image_to_volume(context, volume, image_service, 547 image_id): 548 raise exception.ImageCopyFailure() 549 550 self.mock_object(self.volume.driver, 'copy_image_to_volume', 551 fake_copy_image_to_volume) 552 mock_delete = self.mock_object(self.volume.driver, 'delete_volume') 553 self.assertRaises(exception.ImageCopyFailure, 554 self._create_volume_from_image) 555 # NOTE(dulek): Rescheduling should not occur, so lets assert that 556 # allocated_capacity is incremented. 557 self.assertDictEqual(self.volume.stats['pools'], 558 {'_pool0': {'allocated_capacity_gb': 1}}) 559 # NOTE(dulek): As we haven't rescheduled, make sure no delete_volume 560 # was called. 561 self.assertFalse(mock_delete.called) 562 563 @mock.patch('cinder.utils.brick_get_connector_properties') 564 @mock.patch('cinder.utils.brick_get_connector') 565 @mock.patch('cinder.volume.driver.BaseVD.secure_file_operations_enabled') 566 @mock.patch('cinder.volume.driver.BaseVD._detach_volume') 567 @mock.patch('cinder.image.image_utils.qemu_img_info') 568 def test_create_volume_from_image_unavailable( 569 self, mock_qemu_info, mock_detach, mock_secure, *args): 570 """Test create volume with ImageCopyFailure 571 572 We'll raise an exception inside _connect_device after volume has 573 already been attached to confirm that it detaches the volume. 574 """ 575 mock_secure.side_effect = NameError 576 image_info = imageutils.QemuImgInfo() 577 image_info.virtual_size = '1073741824' 578 mock_qemu_info.return_value = image_info 579 580 unbound_copy_method = cinder.volume.driver.BaseVD.copy_image_to_volume 581 bound_copy_method = unbound_copy_method.__get__(self.volume.driver) 582 with mock.patch.object(self.volume.driver, 'copy_image_to_volume', 583 side_effect=bound_copy_method): 584 self.assertRaises(exception.ImageCopyFailure, 585 self._create_volume_from_image, 586 fakeout_copy_image_to_volume=False) 587 # We must have called detach method. 588 self.assertEqual(1, mock_detach.call_count) 589 590 @mock.patch('cinder.utils.brick_get_connector_properties') 591 @mock.patch('cinder.utils.brick_get_connector') 592 @mock.patch('cinder.volume.driver.BaseVD._connect_device') 593 @mock.patch('cinder.volume.driver.BaseVD._detach_volume') 594 @mock.patch('cinder.image.image_utils.qemu_img_info') 595 def test_create_volume_from_image_unavailable_no_attach_info( 596 self, mock_qemu_info, mock_detach, mock_connect, *args): 597 """Test create volume with ImageCopyFailure 598 599 We'll raise an exception on _connect_device call to confirm that it 600 detaches the volume even if the exception doesn't have attach_info. 601 """ 602 mock_connect.side_effect = NameError 603 image_info = imageutils.QemuImgInfo() 604 image_info.virtual_size = '1073741824' 605 mock_qemu_info.return_value = image_info 606 607 unbound_copy_method = cinder.volume.driver.BaseVD.copy_image_to_volume 608 bound_copy_method = unbound_copy_method.__get__(self.volume.driver) 609 with mock.patch.object(self.volume.driver, 'copy_image_to_volume', 610 side_effect=bound_copy_method): 611 self.assertRaises(exception.ImageCopyFailure, 612 self._create_volume_from_image, 613 fakeout_copy_image_to_volume=False) 614 # We must have called detach method. 615 self.assertEqual(1, mock_detach.call_count) 616 617 @mock.patch('cinder.image.image_utils.qemu_img_info') 618 def test_create_volume_from_image_clone_image_volume(self, mock_qemu_info): 619 """Test create volume from image via image volume. 620 621 Verify that after cloning image to volume, it is in available 622 state and is bootable. 623 """ 624 image_info = imageutils.QemuImgInfo() 625 image_info.virtual_size = '1073741824' 626 mock_qemu_info.return_value = image_info 627 628 volume = self._create_volume_from_image(clone_image_volume=True) 629 self.assertEqual('available', volume['status']) 630 self.assertTrue(volume['bootable']) 631 self.volume.delete_volume(self.context, volume) 632 633 def test_create_volume_from_exact_sized_image(self): 634 """Test create volume from an image of the same size. 635 636 Verify that an image which is exactly the same size as the 637 volume, will work correctly. 638 """ 639 try: 640 volume_id = None 641 volume_api = cinder.volume.api.API( 642 image_service=FakeImageService()) 643 volume = volume_api.create(self.context, 2, 'name', 'description', 644 image_id=self.FAKE_UUID) 645 volume_id = volume['id'] 646 self.assertEqual('creating', volume['status']) 647 648 finally: 649 # cleanup 650 db.volume_destroy(self.context, volume_id) 651 652 def test_create_volume_from_oversized_image(self): 653 """Verify that an image which is too big will fail correctly.""" 654 class _ModifiedFakeImageService(FakeImageService): 655 def show(self, context, image_id): 656 return {'size': 2 * units.Gi + 1, 657 'disk_format': 'raw', 658 'container_format': 'bare', 659 'status': 'active'} 660 661 volume_api = cinder.volume.api.API( 662 image_service=_ModifiedFakeImageService()) 663 664 self.assertRaises(exception.InvalidInput, 665 volume_api.create, 666 self.context, 2, 667 'name', 'description', image_id=1) 668 669 def test_create_volume_with_mindisk_error(self): 670 """Verify volumes smaller than image minDisk will cause an error.""" 671 class _ModifiedFakeImageService(FakeImageService): 672 def show(self, context, image_id): 673 return {'size': 2 * units.Gi, 674 'disk_format': 'raw', 675 'container_format': 'bare', 676 'min_disk': 5, 677 'status': 'active'} 678 679 volume_api = cinder.volume.api.API( 680 image_service=_ModifiedFakeImageService()) 681 682 self.assertRaises(exception.InvalidInput, 683 volume_api.create, 684 self.context, 2, 685 'name', 'description', image_id=1) 686 687 def test_create_volume_with_deleted_imaged(self): 688 """Verify create volume from image will cause an error.""" 689 class _ModifiedFakeImageService(FakeImageService): 690 def show(self, context, image_id): 691 return {'size': 2 * units.Gi, 692 'disk_format': 'raw', 693 'container_format': 'bare', 694 'min_disk': 5, 695 'status': 'deleted'} 696 697 volume_api = cinder.volume.api.API( 698 image_service=_ModifiedFakeImageService()) 699 700 self.assertRaises(exception.InvalidInput, 701 volume_api.create, 702 self.context, 2, 703 'name', 'description', image_id=1) 704 705 def test_copy_volume_to_image_maintenance(self): 706 """Test copy volume to image in maintenance.""" 707 test_meta1 = {'fake_key1': 'fake_value1', 'fake_key2': 'fake_value2'} 708 volume = tests_utils.create_volume(self.context, metadata=test_meta1, 709 **self.volume_params) 710 volume['status'] = 'maintenance' 711 volume_api = cinder.volume.api.API() 712 self.assertRaises(exception.InvalidVolume, 713 volume_api.copy_volume_to_image, 714 self.context, 715 volume, 716 test_meta1, 717 force=True) 718