x/gantt
Code Issues Proposed changes

volume/api: an unit test for create_snapshot_force()

Browse Source
This commit is contained in:
Isaku Yamahata
2011-06-23 19:51:00 +09:00
parent 451e0bdb2d
commit 234d6a56c9
1 changed files with 31 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
nova/tests/test_volume.py
View File

@@ -27,8 +27,10 @@ from nova import exception
from nova import db from nova import db
from nova import flags from nova import flags
from nova import log as logging from nova import log as logging
from nova import rpc
from nova import test from nova import test
from nova import utils from nova import utils
from nova import volume
FLAGS = flags.FLAGS FLAGS = flags.FLAGS
LOG = logging.getLogger('nova.tests.volume') LOG = logging.getLogger('nova.tests.volume')
@@ -43,6 +45,11 @@ class VolumeTestCase(test.TestCase):
self.flags(connection_type='fake') self.flags(connection_type='fake')
self.volume = utils.import_object(FLAGS.volume_manager) self.volume = utils.import_object(FLAGS.volume_manager)
self.context = context.get_admin_context() 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 @staticmethod
def _create_volume(size='0', snapshot_id=None): def _create_volume(size='0', snapshot_id=None):
@@ -224,6 +231,30 @@ class VolumeTestCase(test.TestCase):
snapshot_id) snapshot_id)
self.volume.delete_volume(self.context, volume_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): class DriverTestCase(test.TestCase):
"""Base Test class for Drivers.""" """Base Test class for Drivers."""