Adds create_from_volume test cases

  For tests.test_volumes,
  added 3 test cases:
     create_from_volume
     create_from_volume_with_glance_metadata
     create_from_volume_failed_on_cloning
  refactored an internal api:
     _create_volume_from_image

  The reason not to test a non-exist volume as source_vol is
  the uplayer(volume api) has already checked and ensured the
  source_vol exists

Change-Id: I9c345bc29cae31f1a22236a4849f13d7289ea49e

diff --git a/cinder/tests/test_volume.py b/cinder/tests/test_volume.py
index 75128a43874a643ca467520d71633fb05498beb6..6959c52d104f6229f370cb07245c48e0c0cc4175 100644 (file)
--- a/cinder/tests/test_volume.py
+++ b/cinder/tests/test_volume.py
@@ -79,12 +79,13 @@ class VolumeTestCase(test.TestCase):
 
     @staticmethod
     def _create_volume(size=0, snapshot_id=None, image_id=None,
-                       metadata=None, status="creating"):
+                       source_volid=None, metadata=None, status="creating"):
         """Create a volume object."""
         vol = {}
         vol['size'] = size
         vol['snapshot_id'] = snapshot_id
         vol['image_id'] = image_id
+        vol['source_volid'] = source_volid
         vol['user_id'] = 'fake'
         vol['project_id'] = 'fake'
         vol['availability_zone'] = CONF.storage_availability_zone
@@ -672,8 +673,7 @@ class VolumeTestCase(test.TestCase):
         self.volume.delete_snapshot(self.context, snapshot_id)
         self.volume.delete_volume(self.context, volume_id)
 
-    def _create_volume_from_image(self, expected_status,
-                                  fakeout_copy_image_to_volume=False):
+    def _create_volume_from_image(self, fakeout_copy_image_to_volume=False):
         """Call copy image to volume, Test the status of volume after calling
         copying image to volume.
         """
@@ -696,33 +696,25 @@ class VolumeTestCase(test.TestCase):
                            fake_copy_image_to_volume)
 
         image_id = 'c905cedb-7281-47e4-8a62-f26bc5fc4c77'
-        volume_id = 1
+        volume_id = self._create_volume(status='creating')['id']
         # 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)
+            volume = db.volume_get(self.context, volume_id)
+            return volume
 
     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')
+        volume = self._create_volume_from_image()
+        self.assertEqual(volume['status'], 'available')
+        self.volume.delete_volume(self.context, volume['id'])
 
     def test_create_volume_from_image_exception(self):
         """Verify that create volume from image, the volume status is
@@ -1213,6 +1205,74 @@ class VolumeTestCase(test.TestCase):
         # clean up
         self.volume.delete_volume(self.context, volume['id'])
 
+    def test_create_volume_from_sourcevol(self):
+        """Test volume can be created from a source volume."""
+        def fake_create_cloned_volume(volume, src_vref):
+            pass
+
+        self.stubs.Set(self.volume.driver, 'create_cloned_volume',
+                       fake_create_cloned_volume)
+        volume_src = self._create_volume()
+        self.volume.create_volume(self.context, volume_src['id'])
+        volume_dst = self._create_volume(source_volid=volume_src['id'])
+        self.volume.create_volume(self.context, volume_dst['id'],
+                                  source_volid=volume_src['id'])
+        self.assertEqual('available',
+                         db.volume_get(context.get_admin_context(),
+                                       volume_dst['id']).status)
+        self.volume.delete_volume(self.context, volume_dst['id'])
+        self.volume.delete_volume(self.context, volume_src['id'])
+
+    def test_create_volume_from_sourcevol_with_glance_metadata(self):
+        """Test glance metadata can be correctly copied to new volume."""
+        def fake_create_cloned_volume(volume, src_vref):
+            pass
+
+        self.stubs.Set(self.volume.driver, 'create_cloned_volume',
+                       fake_create_cloned_volume)
+        volume_src = self._create_volume_from_image()
+        self.volume.create_volume(self.context, volume_src['id'])
+        volume_dst = self._create_volume(source_volid=volume_src['id'])
+        self.volume.create_volume(self.context, volume_dst['id'],
+                                  source_volid=volume_src['id'])
+        self.assertEqual('available',
+                         db.volume_get(context.get_admin_context(),
+                                       volume_dst['id']).status)
+        src_glancemeta = db.volume_get(context.get_admin_context(),
+                                       volume_src['id']).volume_glance_metadata
+        dst_glancemeta = db.volume_get(context.get_admin_context(),
+                                       volume_dst['id']).volume_glance_metadata
+        for meta_src in src_glancemeta:
+            for meta_dst in dst_glancemeta:
+                if meta_dst.key == meta_src.key:
+                    self.assertEquals(meta_dst.value, meta_src.value)
+        self.volume.delete_volume(self.context, volume_src['id'])
+        self.volume.delete_volume(self.context, volume_dst['id'])
+
+    def test_create_volume_from_sourcevol_failed_clone(self):
+        """Test src vol status will be restore by error handling code."""
+        def fake_error_create_cloned_volume(volume, src_vref):
+            db.volume_update(context, src_vref['id'], {'status': 'error'})
+            raise exception.CinderException('fake exception')
+
+        def fake_reschedule_or_reraise(context, volume_id, exc_info,
+                                       snapshot_id, image_id, request_spec,
+                                       filter_properties, allow_reschedule):
+            pass
+
+        self.stubs.Set(self.volume, '_reschedule_or_reraise',
+                       fake_reschedule_or_reraise)
+        self.stubs.Set(self.volume.driver, 'create_cloned_volume',
+                       fake_error_create_cloned_volume)
+        volume_src = self._create_volume()
+        self.volume.create_volume(self.context, volume_src['id'])
+        volume_dst = self._create_volume(0, source_volid=volume_src['id'])
+        self.volume.create_volume(self.context, volume_dst['id'],
+                                  source_volid=volume_src['id'])
+        self.assertEqual(volume_src['status'], 'creating')
+        self.volume.delete_volume(self.context, volume_dst['id'])
+        self.volume.delete_volume(self.context, volume_src['id'])
+
 
 class DriverTestCase(test.TestCase):
     """Base Test class for Drivers."""