volume/api: an unit test for create_snapshot_force()

nova/tests/test_volume.py

@@ -27,8 +27,10 @@ from nova import exception
 from nova import db
 from nova import flags
 from nova import log as logging
+from nova import rpc
 from nova import test
 from nova import utils
+from nova import volume
 
 FLAGS = flags.FLAGS
 LOG = logging.getLogger('nova.tests.volume')
@@ -43,6 +45,11 @@ class VolumeTestCase(test.TestCase):
         self.flags(connection_type='fake')
         self.volume = utils.import_object(FLAGS.volume_manager)
         self.context = context.get_admin_context()
+        self.instance_id = db.instance_create(self.context, {})['id']
+
+    def tearDown(self):
+        db.instance_destroy(self.context, self.instance_id)
+        super(VolumeTestCase, self).tearDown()
 
     @staticmethod
     def _create_volume(size='0', snapshot_id=None):
@@ -224,6 +231,30 @@ class VolumeTestCase(test.TestCase):
                           snapshot_id)
         self.volume.delete_volume(self.context, volume_id)
 
+    def test_create_snapshot_force(self):
+        """Test snapshot in use can be created forcibly."""
+
+        def fake_cast(ctxt, topic, msg):
+            pass
+        self.stubs.Set(rpc, 'cast', fake_cast)
+
+        volume_id = self._create_volume()
+        self.volume.create_volume(self.context, volume_id)
+        db.volume_attached(self.context, volume_id, self.instance_id,
+                           '/dev/sda1')
+
+        volume_api = volume.api.API()
+        self.assertRaises(exception.ApiError,
+                          volume_api.create_snapshot,
+                          self.context, volume_id,
+                          'fake_name', 'fake_description')
+        snapshot_ref = volume_api.create_snapshot_force(self.context,
+                                                        volume_id,
+                                                        'fake_name',
+                                                        'fake_description')
+        db.snapshot_destroy(self.context, snapshot_ref['id'])
+        db.volume_destroy(self.context, volume_id)
+
 
 class DriverTestCase(test.TestCase):
     """Base Test class for Drivers."""