diff options
Diffstat (limited to 'nova/tests/test_volume.py')
| -rw-r--r-- | nova/tests/test_volume.py | 241 |
1 files changed, 237 insertions, 4 deletions
diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py index 6de3d380e..0c5328456 100644 --- a/nova/tests/test_volume.py +++ b/nova/tests/test_volume.py @@ -21,8 +21,10 @@ Tests for Volume Code. """ import cStringIO +import datetime import mox +import os import shutil import tempfile @@ -37,9 +39,13 @@ from nova.openstack.common import rpc import nova.policy from nova import quota from nova import test +from nova.tests.image import fake as fake_image +import nova.volume.api from nova.volume import iscsi QUOTAS = quota.QUOTAS + + FLAGS = flags.FLAGS @@ -60,11 +66,12 @@ class VolumeTestCase(test.TestCase): self.instance_id = instance['id'] self.instance_uuid = instance['uuid'] test_notifier.NOTIFICATIONS = [] + fake_image.stub_out_image_service(self.stubs) def tearDown(self): try: shutil.rmtree(FLAGS.volumes_dir) - except OSError, e: + except OSError: pass db.instance_destroy(self.context, self.instance_uuid) notifier_api._reset_drivers() @@ -74,11 +81,12 @@ class VolumeTestCase(test.TestCase): return 1 @staticmethod - def _create_volume(size=0, snapshot_id=None, metadata=None): + def _create_volume(size=0, snapshot_id=None, image_id=None, metadata=None): """Create a volume object.""" vol = {} vol['size'] = size vol['snapshot_id'] = snapshot_id + vol['image_id'] = image_id vol['user_id'] = 'fake' vol['project_id'] = 'fake' vol['availability_zone'] = FLAGS.storage_availability_zone @@ -137,7 +145,7 @@ class VolumeTestCase(test.TestCase): def test_create_delete_volume_with_metadata(self): """Test volume can be created and deleted.""" test_meta = {'fake_key': 'fake_value'} - volume = self._create_volume('0', None, test_meta) + volume = self._create_volume('0', None, metadata=test_meta) volume_id = volume['id'] self.volume.create_volume(self.context, volume_id) result_meta = { @@ -564,6 +572,231 @@ class VolumeTestCase(test.TestCase): volume = db.volume_get(self.context, volume['id']) self.assertEqual(volume['status'], "in-use") + def _create_volume_from_image(self, expected_status, + fakeout_copy_image_to_volume=False): + """Call copy image to volume, Test the status of volume after calling + copying image to volume.""" + def fake_local_path(volume): + return dst_path + + def fake_copy_image_to_volume(context, volume, image_id): + pass + + dst_fd, dst_path = tempfile.mkstemp() + os.close(dst_fd) + self.stubs.Set(self.volume.driver, 'local_path', fake_local_path) + if fakeout_copy_image_to_volume: + self.stubs.Set(self.volume, '_copy_image_to_volume', + fake_copy_image_to_volume) + + image_id = 'c905cedb-7281-47e4-8a62-f26bc5fc4c77' + volume_id = 1 + # creating volume testdata + db.volume_create(self.context, {'id': volume_id, + 'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1), + 'display_description': 'Test Desc', + 'size': 20, + 'status': 'creating', + 'instance_uuid': None, + 'host': 'dummy'}) + try: + self.volume.create_volume(self.context, + volume_id, + image_id=image_id) + + volume = db.volume_get(self.context, volume_id) + self.assertEqual(volume['status'], expected_status) + finally: + # cleanup + db.volume_destroy(self.context, volume_id) + os.unlink(dst_path) + + def test_create_volume_from_image_status_downloading(self): + """Verify that before copying image to volume, it is in downloading + state.""" + self._create_volume_from_image('downloading', True) + + def test_create_volume_from_image_status_available(self): + """Verify that before copying image to volume, it is in available + state.""" + self._create_volume_from_image('available') + + def test_create_volume_from_image_exception(self): + """Verify that create volume from image, the volume status is + 'downloading'.""" + dst_fd, dst_path = tempfile.mkstemp() + os.close(dst_fd) + + self.stubs.Set(self.volume.driver, 'local_path', lambda x: dst_path) + + image_id = 'aaaaaaaa-0000-0000-0000-000000000000' + # creating volume testdata + volume_id = 1 + db.volume_create(self.context, {'id': volume_id, + 'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1), + 'display_description': 'Test Desc', + 'size': 20, + 'status': 'creating', + 'host': 'dummy'}) + + self.assertRaises(exception.ImageNotFound, + self.volume.create_volume, + self.context, + volume_id, + None, + image_id) + volume = db.volume_get(self.context, volume_id) + self.assertEqual(volume['status'], "error") + # cleanup + db.volume_destroy(self.context, volume_id) + os.unlink(dst_path) + + def test_copy_volume_to_image_status_available(self): + dst_fd, dst_path = tempfile.mkstemp() + os.close(dst_fd) + + def fake_local_path(volume): + return dst_path + + self.stubs.Set(self.volume.driver, 'local_path', fake_local_path) + + image_id = '70a599e0-31e7-49b7-b260-868f441e862b' + # creating volume testdata + volume_id = 1 + db.volume_create(self.context, {'id': volume_id, + 'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1), + 'display_description': 'Test Desc', + 'size': 20, + 'status': 'uploading', + 'instance_uuid': None, + 'host': 'dummy'}) + + try: + # start test + self.volume.copy_volume_to_image(self.context, + volume_id, + image_id) + + volume = db.volume_get(self.context, volume_id) + self.assertEqual(volume['status'], 'available') + finally: + # cleanup + db.volume_destroy(self.context, volume_id) + os.unlink(dst_path) + + def test_copy_volume_to_image_status_use(self): + dst_fd, dst_path = tempfile.mkstemp() + os.close(dst_fd) + + def fake_local_path(volume): + return dst_path + + self.stubs.Set(self.volume.driver, 'local_path', fake_local_path) + + #image_id = '70a599e0-31e7-49b7-b260-868f441e862b' + image_id = 'a440c04b-79fa-479c-bed1-0b816eaec379' + # creating volume testdata + volume_id = 1 + db.volume_create(self.context, + {'id': volume_id, + 'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1), + 'display_description': 'Test Desc', + 'size': 20, + 'status': 'uploading', + 'instance_uuid': + 'b21f957d-a72f-4b93-b5a5-45b1161abb02', + 'host': 'dummy'}) + + try: + # start test + self.volume.copy_volume_to_image(self.context, + volume_id, + image_id) + + volume = db.volume_get(self.context, volume_id) + self.assertEqual(volume['status'], 'in-use') + finally: + # cleanup + db.volume_destroy(self.context, volume_id) + os.unlink(dst_path) + + def test_copy_volume_to_image_exception(self): + dst_fd, dst_path = tempfile.mkstemp() + os.close(dst_fd) + + def fake_local_path(volume): + return dst_path + + self.stubs.Set(self.volume.driver, 'local_path', fake_local_path) + + image_id = 'aaaaaaaa-0000-0000-0000-000000000000' + # creating volume testdata + volume_id = 1 + db.volume_create(self.context, {'id': volume_id, + 'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1), + 'display_description': 'Test Desc', + 'size': 20, + 'status': 'in-use', + 'host': 'dummy'}) + + try: + # start test + self.assertRaises(exception.ImageNotFound, + self.volume.copy_volume_to_image, + self.context, + volume_id, + image_id) + + volume = db.volume_get(self.context, volume_id) + self.assertEqual(volume['status'], 'available') + finally: + # cleanup + db.volume_destroy(self.context, volume_id) + os.unlink(dst_path) + + def test_create_volume_from_exact_sized_image(self): + """Verify that an image which is exactly the same size as the + volume, will work correctly.""" + class _FakeImageService: + def __init__(self, db_driver=None, image_service=None): + pass + + def show(self, context, image_id): + return {'size': 2 * 1024 * 1024 * 1024} + + image_id = '70a599e0-31e7-49b7-b260-868f441e862b' + + try: + volume_id = None + volume_api = nova.volume.api.API( + image_service=_FakeImageService()) + volume = volume_api.create(self.context, 2, 'name', 'description', + image_id=1) + volume_id = volume['id'] + self.assertEqual(volume['status'], 'creating') + + finally: + # cleanup + db.volume_destroy(self.context, volume_id) + + def test_create_volume_from_oversized_image(self): + """Verify that an image which is too big will fail correctly.""" + class _FakeImageService: + def __init__(self, db_driver=None, image_service=None): + pass + + def show(self, context, image_id): + return {'size': 2 * 1024 * 1024 * 1024 + 1} + + image_id = '70a599e0-31e7-49b7-b260-868f441e862b' + + volume_api = nova.volume.api.API(image_service=_FakeImageService()) + + self.assertRaises(exception.InvalidInput, + volume_api.create, + self.context, 2, + 'name', 'description', image_id=1) + class DriverTestCase(test.TestCase): """Base Test class for Drivers.""" @@ -591,7 +824,7 @@ class DriverTestCase(test.TestCase): def tearDown(self): try: shutil.rmtree(FLAGS.volumes_dir) - except OSError, e: + except OSError: pass super(DriverTestCase, self).tearDown() |