from cinder import exception
from cinder.i18n import _, _LE, _LW
from cinder import objects
+from cinder.objects import fields as c_fields
import cinder.policy
from cinder import quota
from cinder.scheduler import rpcapi as scheduler_rpcapi
LOG = logging.getLogger(__name__)
CGQUOTAS = quota.CGQUOTAS
-VALID_REMOVE_VOL_FROM_CG_STATUS = ('available', 'in-use',)
+VALID_REMOVE_VOL_FROM_CG_STATUS = (c_fields.ConsistencyGroupStatus.AVAILABLE,
+ c_fields.ConsistencyGroupStatus.IN_USE)
def wrap_check_policy(func):
kwargs = {'user_id': context.user_id,
'project_id': context.project_id,
'availability_zone': availability_zone,
- 'status': "creating",
+ 'status': c_fields.ConsistencyGroupStatus.CREATING,
'name': name,
'description': description,
'volume_type_id': req_volume_type_ids}
kwargs = {
'user_id': context.user_id,
'project_id': context.project_id,
- 'status': "creating",
+ 'status': c_fields.ConsistencyGroupStatus.CREATING,
'name': name,
'description': description,
'cgsnapshot_id': cgsnapshot_id,
return
- if not force and group.status not in ["available", "error"]:
+ if not force and group.status not in (
+ [c_fields.ConsistencyGroupStatus.AVAILABLE,
+ c_fields.ConsistencyGroupStatus.ERROR]):
msg = _("Consistency group status must be available or error, "
"but current status is: %s") % group.status
raise exception.InvalidConsistencyGroup(reason=msg)
LOG.error(msg)
raise exception.InvalidConsistencyGroup(reason=msg)
- group.status = 'deleting'
+ group.status = c_fields.ConsistencyGroupStatus.DELETING
group.terminated_at = timeutils.utcnow()
group.save()
def update(self, context, group, name, description,
add_volumes, remove_volumes):
"""Update consistency group."""
- if group.status != 'available':
+ if group.status != c_fields.ConsistencyGroupStatus.AVAILABLE:
msg = _("Consistency group status must be available, "
"but current status is: %s.") % group.status
raise exception.InvalidConsistencyGroup(reason=msg)
with session.begin():
model_query(context, models.ConsistencyGroup, session=session).\
filter_by(id=consistencygroup_id).\
- update({'status': 'deleted',
+ update({'status': fields.ConsistencyGroupStatus.DELETED,
'deleted': True,
'deleted_at': timeutils.utcnow(),
'updated_at': literal_column('updated_at')})
from cinder.i18n import _
from cinder import objects
from cinder.objects import base
+from cinder.objects import fields as c_fields
from oslo_versionedobjects import fields
OPTIONAL_FIELDS = ['cgsnapshots', 'volumes']
base.CinderObjectDictCompat):
# Version 1.0: Initial version
# Version 1.1: Added cgsnapshots and volumes relationships
- VERSION = '1.1'
+ # Version 1.2: Changed 'status' field to use ConsistencyGroupStatusField
+ VERSION = '1.2'
fields = {
'id': fields.UUIDField(),
'name': fields.StringField(nullable=True),
'description': fields.StringField(nullable=True),
'volume_type_id': fields.UUIDField(nullable=True),
- 'status': fields.StringField(nullable=True),
+ 'status': c_fields.ConsistencyGroupStatusField(nullable=True),
'cgsnapshot_id': fields.UUIDField(nullable=True),
'source_cgid': fields.UUIDField(nullable=True),
'cgsnapshots': fields.ObjectField('CGSnapshotList', nullable=True),
class BackupStatusField(BaseEnumField):
AUTO_TYPE = BackupStatus()
+
+
+class ConsistencyGroupStatus(Enum):
+ ERROR = 'error'
+ AVAILABLE = 'available'
+ CREATING = 'creating'
+ DELETING = 'deleting'
+ DELETED = 'deleted'
+ UPDATING = 'updating'
+ IN_USE = 'in-use'
+ ERROR_DELETING = 'error_deleting'
+
+ ALL = (ERROR, AVAILABLE, CREATING, DELETING, DELETED,
+ UPDATING, IN_USE, ERROR_DELETING)
+
+ def __init__(self):
+ super(ConsistencyGroupStatus, self).__init__(
+ valid_values=ConsistencyGroupStatus.ALL)
+
+
+class ConsistencyGroupStatusField(BaseEnumField):
+ AUTO_TYPE = ConsistencyGroupStatus()
from cinder import exception
from cinder.i18n import _
from cinder import objects
+from cinder.objects import fields
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit.api.v2 import stubs
volume_type_id='123456',
availability_zone='az1',
host='fakehost',
- status='creating'):
+ status=fields.ConsistencyGroupStatus.CREATING):
"""Create a consistency group object."""
ctxt = ctxt or self.ctxt
consistencygroup = objects.ConsistencyGroup(ctxt)
res_dict['badRequest']['message'])
def test_delete_consistencygroup_available(self):
- consistencygroup = self._create_consistencygroup(status='available')
+ consistencygroup = self._create_consistencygroup(
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/delete' %
consistencygroup.id)
req.method = 'POST'
consistencygroup = objects.ConsistencyGroup.get_by_id(
self.ctxt, consistencygroup.id)
self.assertEqual(202, res.status_int)
- self.assertEqual('deleting', consistencygroup.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETING,
+ consistencygroup.status)
consistencygroup.destroy()
res_dict['itemNotFound']['message'])
def test_delete_consistencygroup_with_Invalidconsistencygroup(self):
- consistencygroup = self._create_consistencygroup(status='invalid')
+ consistencygroup = self._create_consistencygroup(
+ status=fields.ConsistencyGroupStatus.IN_USE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/delete' %
consistencygroup.id)
req.method = 'POST'
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
msg = (_('Invalid ConsistencyGroup: Consistency group status must be '
- 'available or error, but current status is: invalid'))
+ 'available or error, but current status is: in-use'))
self.assertEqual(msg, res_dict['badRequest']['message'])
consistencygroup.destroy()
def test_delete_consistencygroup_no_host(self):
consistencygroup = self._create_consistencygroup(
host=None,
- status='error')
+ status=fields.ConsistencyGroupStatus.ERROR)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/delete' %
consistencygroup.id)
req.method = 'POST'
cg = objects.ConsistencyGroup.get_by_id(
context.get_admin_context(read_deleted='yes'),
consistencygroup.id)
- self.assertEqual('deleted', cg.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED, cg.status)
self.assertIsNone(cg.host)
def test_create_delete_consistencygroup_update_quota(self):
fake_type['name'])
self.cg_api.update_quota.assert_called_once_with(
self.ctxt, cg, 1)
- self.assertEqual('creating', cg.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.CREATING, cg.status)
self.assertIsNone(cg.host)
self.cg_api.update_quota.reset_mock()
- cg.status = 'error'
+ cg.status = fields.ConsistencyGroupStatus.ERROR
self.cg_api.delete(self.ctxt, cg)
self.cg_api.update_quota.assert_called_once_with(
self.ctxt, cg, -1, self.ctxt.project_id)
cg = objects.ConsistencyGroup.get_by_id(
context.get_admin_context(read_deleted='yes'),
cg.id)
- self.assertEqual('deleted', cg.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED, cg.status)
def test_delete_consistencygroup_with_invalid_body(self):
- consistencygroup = self._create_consistencygroup(status='available')
+ consistencygroup = self._create_consistencygroup(
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/delete' %
consistencygroup.id)
req.method = 'POST'
self.assertEqual(400, res.status_int)
def test_delete_consistencygroup_with_invalid_force_value_in_body(self):
- consistencygroup = self._create_consistencygroup(status='available')
+ consistencygroup = self._create_consistencygroup(
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/delete' %
consistencygroup.id)
req.method = 'POST'
self.assertEqual(400, res.status_int)
def test_delete_consistencygroup_with_empty_force_value_in_body(self):
- consistencygroup = self._create_consistencygroup(status='available')
+ consistencygroup = self._create_consistencygroup(
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/delete' %
consistencygroup.id)
req.method = 'POST'
'cinder.api.openstack.wsgi.Controller.validate_name_and_description')
def test_update_consistencygroup_success(self, mock_validate):
volume_type_id = '123456'
- consistencygroup = self._create_consistencygroup(status='available',
- host='test_host')
+ consistencygroup = self._create_consistencygroup(
+ status=fields.ConsistencyGroupStatus.AVAILABLE,
+ host='test_host')
remove_volume_id = utils.create_volume(
self.ctxt,
volume_type_id=volume_type_id,
consistencygroup_id=consistencygroup.id)['id']
- self.assertEqual('available', consistencygroup.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ consistencygroup.status)
cg_volumes = db.volume_get_all_by_group(self.ctxt.elevated(),
consistencygroup.id)
self.ctxt, consistencygroup.id)
self.assertEqual(202, res.status_int)
self.assertTrue(mock_validate.called)
- self.assertEqual('updating', consistencygroup.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.UPDATING,
+ consistencygroup.status)
consistencygroup.destroy()
def test_update_consistencygroup_add_volume_not_found(self):
- consistencygroup = self._create_consistencygroup(ctxt=self.ctxt,
- status='available')
+ consistencygroup = self._create_consistencygroup(
+ ctxt=self.ctxt,
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/update' %
consistencygroup.id)
req.method = 'PUT'
consistencygroup.destroy()
def test_update_consistencygroup_remove_volume_not_found(self):
- consistencygroup = self._create_consistencygroup(ctxt=self.ctxt,
- status='available')
+ consistencygroup = self._create_consistencygroup(
+ ctxt=self.ctxt,
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/update' %
consistencygroup.id)
req.method = 'PUT'
consistencygroup.destroy()
def test_update_consistencygroup_empty_parameters(self):
- consistencygroup = self._create_consistencygroup(ctxt=self.ctxt,
- status='available')
+ consistencygroup = self._create_consistencygroup(
+ ctxt=self.ctxt,
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/update' %
consistencygroup.id)
req.method = 'PUT'
def test_update_consistencygroup_add_volume_invalid_state(self):
volume_type_id = '123456'
- consistencygroup = self._create_consistencygroup(ctxt=self.ctxt,
- status='available')
+ consistencygroup = self._create_consistencygroup(
+ ctxt=self.ctxt,
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
add_volume_id = utils.create_volume(
self.ctxt,
volume_type_id=volume_type_id,
consistencygroup.destroy()
def test_update_consistencygroup_add_volume_invalid_volume_type(self):
- consistencygroup = self._create_consistencygroup(ctxt=self.ctxt,
- status='available')
+ consistencygroup = self._create_consistencygroup(
+ ctxt=self.ctxt,
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
wrong_type = 'wrong-volume-type-id'
add_volume_id = utils.create_volume(
self.ctxt,
consistencygroup.destroy()
def test_update_consistencygroup_add_volume_already_in_cg(self):
- consistencygroup = self._create_consistencygroup(ctxt=self.ctxt,
- status='available')
+ consistencygroup = self._create_consistencygroup(
+ ctxt=self.ctxt,
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
add_volume_id = utils.create_volume(
self.ctxt,
consistencygroup_id='some_other_cg')['id']
consistencygroup.destroy()
def test_update_consistencygroup_invalid_state(self):
- wrong_status = 'wrong_status'
- consistencygroup = self._create_consistencygroup(status=wrong_status,
- ctxt=self.ctxt)
+ consistencygroup = self._create_consistencygroup(
+ status=fields.ConsistencyGroupStatus.IN_USE,
+ ctxt=self.ctxt)
req = webob.Request.blank('/v2/fake/consistencygroups/%s/update' %
consistencygroup.id)
req.method = 'PUT'
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
msg = _("Invalid ConsistencyGroup: Consistency group status must be "
- "available, but current status is: %s.") % wrong_status
+ "available, but current status is: %s.") % (
+ fields.ConsistencyGroupStatus.IN_USE)
self.assertEqual(msg, res_dict['badRequest']['message'])
consistencygroup.destroy()
from oslo_utils import timeutils
from cinder import exception
+from cinder.objects import fields
from cinder.tests.unit.brick import fake_lvm
from cinder.volume import driver
from cinder.volume.drivers import lvm
# A consistencygroup entry is already created in db
# This driver just returns a status
now = timeutils.utcnow()
- model_update = {'status': 'available', 'updated_at': now}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE,
+ 'updated_at': now}
return model_update
volume_model_update['status'] = 'available'
except Exception:
volume_model_update['status'] = 'error'
- model_update['status'] = 'error'
+ model_update['status'] = fields.ConsistencyGroupStatus.ERROR
volume_model_updates.append(volume_model_update)
return model_update, volume_model_updates
from cinder import exception
from cinder import objects
+from cinder.objects import fields
from cinder.tests.unit import objects as test_objects
fake_consistencygroup = {
'name': 'fake_name',
'description': 'fake_description',
'volume_type_id': 'fake_volume_type_id',
- 'status': 'creating',
+ 'status': fields.ConsistencyGroupStatus.CREATING,
'cgsnapshot_id': 'fake_id',
'source_cgid': None,
}
def test_save(self, consistencygroup_update):
consistencygroup = objects.ConsistencyGroup._from_db_object(
self.context, objects.ConsistencyGroup(), fake_consistencygroup)
- consistencygroup.status = 'active'
+ consistencygroup.status = fields.ConsistencyGroupStatus.AVAILABLE
consistencygroup.save()
- consistencygroup_update.assert_called_once_with(self.context,
- consistencygroup.id,
- {'status': 'active'})
+ consistencygroup_update.assert_called_once_with(
+ self.context,
+ consistencygroup.id,
+ {'status': fields.ConsistencyGroupStatus.AVAILABLE})
def test_save_with_cgsnapshots(self):
consistencygroup = objects.ConsistencyGroup._from_db_object(
def test_stringify_invalid(self):
self.assertRaises(ValueError, self.field.stringify, 'not_a_status')
+
+
+class TestConsistencyGroupStatus(TestField):
+ def setUp(self):
+ super(TestConsistencyGroupStatus, self).setUp()
+ self.field = fields.ConsistencyGroupStatusField()
+ self.coerce_good_values = [('error', 'error'),
+ ('available', 'available'),
+ ('creating', 'creating'),
+ ('deleting', 'deleting'),
+ ('deleted', 'deleted'),
+ ('updating', 'updating'),
+ ('in-use', 'in-use'),
+ ('error_deleting', 'error_deleting')]
+ self.coerce_bad_values = ['acme']
+ self.to_primitive_values = self.coerce_good_values[0:1]
+ self.from_primitive_values = self.coerce_good_values[0:1]
+
+ def test_stringify(self):
+ self.assertEqual("'error'", self.field.stringify('error'))
+
+ def test_stringify_invalid(self):
+ self.assertRaises(ValueError, self.field.stringify, 'not_a_status')
'BackupList': '1.0-24591dabe26d920ce0756fe64cd5f3aa',
'CGSnapshot': '1.0-190da2a2aa9457edc771d888f7d225c4',
'CGSnapshotList': '1.0-e8c3f4078cd0ee23487b34d173eec776',
- 'ConsistencyGroup': '1.1-8d8b867a67c1bd6e9f840bcf5e375dbb',
+ 'ConsistencyGroup': '1.2-ed7f90a6871991a19af716ade7337fc9',
'ConsistencyGroupList': '1.0-09d0aad5491e762ecfdf66bef02ceb8d',
'Service': '1.0-64baeb4911dbab1153064dd1c87edb9f',
'ServiceList': '1.0-d242d3384b68e5a5a534e090ff1d5161',
from cinder import context
from cinder import db
from cinder import exception
+from cinder.objects import fields
from cinder.scheduler import driver
from cinder.scheduler import filter_scheduler
from cinder.scheduler import manager
consistencygroup_obj)
self.assertTrue(LOG.exception.call_count > 0)
db.consistencygroup_update.assert_called_once_with(
- self.context, group_id, {'status': 'error'})
+ self.context, group_id, {'status': (
+ fields.ConsistencyGroupStatus.ERROR)})
mock_cg.reset_mock()
LOG.exception.reset_mock()
self.context, 'volume', consistencygroup_obj)
self.assertTrue(LOG.error.call_count > 0)
db.consistencygroup_update.assert_called_once_with(
- self.context, group_id, {'status': 'error'})
+ self.context, group_id, {'status': (
+ fields.ConsistencyGroupStatus.ERROR)})
self.manager.driver = original_driver
from cinder import context
from cinder import exception
+from cinder.objects import fields
from cinder import test
from cinder.volume.drivers.dell import dell_storagecenter_api
from cinder.volume.drivers.dell import dell_storagecenter_common
self.driver.db.volume_get_all_by_group.return_value = expected_volumes
context = {}
group = {'id': 'fc8f2fec-fab2-4e34-9148-c094c913b9a3',
- 'status': 'deleted'}
+ 'status': fields.ConsistencyGroupStatus.DELETED}
model_update, volumes = self.driver.delete_consistencygroup(context,
group,
[])
self.driver.db.volume_get_all_by_group.return_value = expected_volumes
context = {}
group = {'id': 'fc8f2fec-fab2-4e34-9148-c094c913b9a3',
- 'status': 'deleted'}
+ 'status': fields.ConsistencyGroupStatus.DELETED}
model_update, volumes = self.driver.delete_consistencygroup(context,
group,
[])
from cinder import exception
from cinder.i18n import _
+from cinder.objects import fields
from cinder import test
from cinder.volume.drivers.emc import emc_vmax_common
from cinder.volume.drivers.emc import emc_vmax_fast
test_CG = {'name': 'myCG1',
'id': '12345abcde',
'volume_type_id': 'abc',
- 'status': 'available'
+ 'status': fields.ConsistencyGroupStatus.AVAILABLE
}
test_snapshot = {'name': 'myCG1',
'id': '12345abcde',
self.driver.create_consistencygroup_from_src(
self.data.test_ctxt, self.data.test_CG, volumes,
self.data.test_CG_snapshot, snapshots))
- self.assertEqual({'status': 'available'}, model_update)
+ self.assertEqual({'status': fields.ConsistencyGroupStatus.AVAILABLE},
+ model_update)
self.assertEqual([{'status': 'available', 'id': '2'}],
volumes_model_update)
from cinder import context
from cinder import exception
+from cinder.objects import fields
from cinder import test
from cinder.tests.unit import fake_consistencygroup
from cinder.tests.unit import fake_snapshot
test_cg = {'id': 'consistencygroup_id',
'name': 'group_name',
- 'status': 'deleting'}
+ 'status': fields.ConsistencyGroupStatus.DELETING}
test_cg_with_type = {'id': 'consistencygroup_id',
'name': 'group_name',
- 'status': 'creating',
+ 'status': fields.ConsistencyGroupStatus.CREATING,
'volume_type_id':
'abc1-2320-9013-8813-8941-1374-8112-1231,'
'19fdd0dd-03b3-4d7c-b541-f4df46f308c8,'}
model_update = self.driver.create_consistencygroup(
None, self.testData.test_cg)
- self.assertDictMatch({'status': 'available'}, model_update)
+ self.assertDictMatch({'status': (
+ fields.ConsistencyGroupStatus.AVAILABLE)}, model_update)
expect_cmd = [
mock.call(
*self.testData.CREATE_CONSISTENCYGROUP_CMD(
fake_cli = self.driverSetup(commands, results)
model_update = self.driver.create_consistencygroup(
None, self.testData.test_cg)
- self.assertDictMatch({'status': 'available'}, model_update)
+ self.assertDictMatch({'status': (
+ fields.ConsistencyGroupStatus.AVAILABLE)}, model_update)
expect_cmd = [
mock.call(
*self.testData.CREATE_CONSISTENCYGROUP_CMD(
mock.call(*self.testData.REPLACE_LUNS_IN_CG_CMD(
cg_name, ['4', '5']), poll=False)]
fake_cli.assert_has_calls(expect_cmd)
- self.assertEqual('available', model_update['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ model_update['status'])
def test_update_consistencygroup_remove_all(self):
cg_name = self.testData.test_cg['id']
mock.call(*self.testData.REMOVE_LUNS_FROM_CG_CMD(
cg_name, ['1', '3']), poll=False)]
fake_cli.assert_has_calls(expect_cmd)
- self.assertEqual('available', model_update['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ model_update['status'])
def test_update_consistencygroup_remove_not_in_cg(self):
cg_name = self.testData.test_cg['id']
mock.call(*self.testData.REPLACE_LUNS_IN_CG_CMD(
cg_name, ['1', '3']), poll=False)]
fake_cli.assert_has_calls(expect_cmd)
- self.assertEqual('available', model_update['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ model_update['status'])
def test_update_consistencygroup_error(self):
cg_name = self.testData.test_cg['id']
from cinder import context
from cinder import exception
+from cinder.objects import fields
from cinder import test
from cinder import utils
from cinder.volume import configuration as conf
def test_delete_consistencygroup(self, mock_exec):
ctxt = self.context
group = self._fake_group()
- group['status'] = 'available'
+ group['status'] = fields.ConsistencyGroupStatus.AVAILABLE
volume = self._fake_volume()
volume['status'] = 'available'
volumes = []
def test_delete_consistencygroup_fail(self, mock_exec):
ctxt = self.context
group = self._fake_group()
- group['status'] = 'available'
+ group['status'] = fields.ConsistencyGroupStatus.AVAILABLE
self.driver.db = mock.Mock()
self.driver.db.volume_get_all_by_group = mock.Mock()
self.driver.db.volume_get_all_by_group.return_value = []
from cinder import context
from cinder import exception
+from cinder.objects import fields
from cinder import test
from cinder.tests.unit import fake_hpe_3par_client as hpe3parclient
from cinder.volume.drivers.hpe import hpe_3par_common as hpecommon
mock_client.reset_mock()
# remove the consistency group
- group.status = 'deleting'
+ group.status = fields.ConsistencyGroupStatus.DELETING
self.driver.delete_consistencygroup(context.get_admin_context(),
group, [])
from cinder import context
from cinder import exception
+from cinder.objects import fields
from cinder import test
from cinder.tests.unit import fake_hpe_lefthand_client as hpelefthandclient
from cinder.volume.drivers.hpe import hpe_lefthand_iscsi
group = self.fake_consistencygroup_object()
cg = self.driver.create_consistencygroup(ctxt, group)
- self.assertEqual('available', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ cg['status'])
def test_delete_consistencygroup(self):
ctxt = context.get_admin_context()
# create a consistency group
group = self.fake_consistencygroup_object()
cg = self.driver.create_consistencygroup(ctxt, group)
- self.assertEqual('available', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ cg['status'])
# delete the consistency group
- group.status = 'deleting'
+ group.status = fields.ConsistencyGroupStatus.DELETING
cg, vols = self.driver.delete_consistencygroup(ctxt, group,
volumes)
- self.assertEqual('deleting', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETING,
+ cg['status'])
def test_update_consistencygroup_add_vol_delete_cg(self):
ctxt = context.get_admin_context()
# create a consistency group
group = self.fake_consistencygroup_object()
cg = self.driver.create_consistencygroup(ctxt, group)
- self.assertEqual('available', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ cg['status'])
# add volume to consistency group
cg = self.driver.update_consistencygroup(
ctxt, group, add_volumes=[self.volume], remove_volumes=None)
# delete the consistency group
- group.status = 'deleting'
+ group.status = fields.ConsistencyGroupStatus.DELETING
cg, vols = self.driver.delete_consistencygroup(ctxt, group,
volumes)
- self.assertEqual('deleting', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETING,
+ cg['status'])
def test_update_consistencygroup_remove_vol_delete_cg(self):
ctxt = context.get_admin_context()
# create a consistency group
group = self.fake_consistencygroup_object()
cg = self.driver.create_consistencygroup(ctxt, group)
- self.assertEqual('available', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ cg['status'])
# add volume to consistency group
cg = self.driver.update_consistencygroup(
ctxt, group, add_volumes=None, remove_volumes=[self.volume])
# delete the consistency group
- group.status = 'deleting'
+ group.status = fields.ConsistencyGroupStatus.DELETING
cg, vols = self.driver.delete_consistencygroup(ctxt, group,
volumes)
- self.assertEqual('deleting', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETING,
+ cg['status'])
def test_create_cgsnapshot(self):
ctxt = context.get_admin_context()
# create a consistency group
group = self.fake_consistencygroup_object()
cg = self.driver.create_consistencygroup(ctxt, group)
- self.assertEqual('available', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ cg['status'])
# create volume and add it to the consistency group
self.driver.update_consistencygroup(
# create a consistency group
group = self.fake_consistencygroup_object()
cg = self.driver.create_consistencygroup(ctxt, group)
- self.assertEqual('available', cg['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
+ cg['status'])
# create volume and add it to the consistency group
self.driver.update_consistencygroup(
from cinder import context
from cinder import exception
from cinder.i18n import _
+from cinder.objects import fields
from cinder import test
from cinder.volume import configuration as conf
from cinder.volume.drivers.ibm import xiv_ds8k
raise exception.CinderException(
message='The consistency group id of volume may be wrong.')
- return {'status': 'available'}
+ return {'status': fields.ConsistencyGroupStatus.AVAILABLE}
def delete_consistencygroup(self, ctxt, group, volumes):
for volume in self.volumes.values():
self, context, group,
add_volumes, remove_volumes):
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
return model_update, None, None
def create_consistencygroup_from_src(
# Create consistency group
model_update = self.driver.create_consistencygroup(ctxt, CONSISTGROUP)
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"Consistency Group created failed")
ctxt, CONSISTGROUP, [CG_VOLUME])
# Verify the result
- self.assertEqual('deleted',
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
model_update['status'],
'Consistency Group deleted failed')
for volume in volumes:
model_update, added, removed = self.driver.update_consistencygroup(
ctxt, CONSISTGROUP, [], [])
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"Consistency Group update failed")
self.assertFalse(added,
model_update, added, removed = self.driver.update_consistencygroup(
ctxt, CONSISTGROUP, [VOLUME], [VOLUME2])
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"Consistency Group update failed")
self.assertFalse(added,
# model_update can be None or return available in status
if model_update:
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"Consistency Group create from source failed")
# volumes_model_update can be None or return available in status
# model_update can be None or return available in status
if model_update:
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"Consistency Group create from source failed")
# volumes_model_update can be None or return available in status
from six.moves import http_client
from cinder import exception
+from cinder.objects import fields
from cinder import test
from cinder.tests.unit import fake_snapshot
from cinder.volume import configuration as conf
self.DPL_MOCK.create_vg.assert_called_once_with(
self._conver_uuid2hex(DATA_IN_GROUP['id']), DATA_IN_GROUP['name'],
DATA_IN_GROUP['description'])
- self.assertDictMatch({'status': 'available'}, model_update)
+ self.assertDictMatch({'status': (
+ fields.ConsistencyGroupStatus.AVAILABLE)}, model_update)
def test_delete_consistency_group(self):
self.DB_MOCK.volume_get_all_by_group.return_value = (
self._conver_uuid2hex(DATA_IN_GROUP['id']))
self.DPL_MOCK.delete_vdev.assert_called_once_with(
self._conver_uuid2hex((DATA_IN_VOLUME_VG['id'])))
- self.assertDictMatch({'status': 'deleted'}, model_update)
+ self.assertDictMatch({'status': (
+ fields.ConsistencyGroupStatus.DELETED)}, model_update)
def test_update_consistencygroup(self):
self.DPL_MOCK.get_vg.return_value = (0, DATA_OUT_CG)
self.DPL_MOCK.leave_vg.assert_called_once_with(
self._conver_uuid2hex(remove_vol['id']),
self._conver_uuid2hex(DATA_IN_GROUP['id']))
- self.assertDictMatch({'status': 'available'}, model_update)
+ self.assertDictMatch({'status': (
+ fields.ConsistencyGroupStatus.AVAILABLE)}, model_update)
def test_update_consistencygroup_exception_join(self):
self.DPL_MOCK.get_vg.return_value = (0, DATA_OUT_CG)
from cinder import context
from cinder import exception
+from cinder.objects import fields
from cinder import test
from cinder.volume import configuration as conf
from cinder.volume.drivers import solidfire
cgsnapshot, snapshots,
source_cg, source_vols)
get_snap.assert_called_with(name)
- self.assertEqual({'status': 'available'}, model)
+ self.assertEqual(
+ {'status': fields.ConsistencyGroupStatus.AVAILABLE}, model)
def test_create_consisgroup_from_src_source_cg(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
source_cg,
source_vols)
get_snap.assert_called_with(source_cg['id'])
- self.assertEqual({'status': 'available'}, model)
+ self.assertEqual(
+ {'status': fields.ConsistencyGroupStatus.AVAILABLE}, model)
def test_create_cgsnapshot(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
from cinder import context
from cinder import exception
from cinder.i18n import _
+from cinder.objects import fields
from cinder import test
from cinder.tests.unit import utils as testutils
from cinder import utils
cg = self._create_consistencygroup_in_db(**kwargs)
model_update = self.driver.create_consistencygroup(self.ctxt, cg)
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"CG created failed")
return cg
model_update = self.driver.create_consistencygroup(self.ctxt, cg)
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"CG created failed")
# Add volumes to CG
model_update = self.driver.delete_consistencygroup(self.ctxt, cg, [])
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for volume in model_update[1]:
self.assertEqual('deleted', volume['status'])
model_update = self.driver.delete_consistencygroup(self.ctxt, cg, [])
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for volume in model_update[1]:
self.assertEqual('deleted', volume['status'])
model_update = (
self.driver.delete_consistencygroup(self.ctxt, source_cg, []))
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for volume in model_update[1]:
self.assertEqual('deleted', volume['status'])
model_update = (
self.driver.delete_consistencygroup(self.ctxt, cgsnapshot, []))
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for volume in model_update[1]:
self.assertEqual('deleted', volume['status'])
None, None,
source_cg,
source_vols))
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"CG create from src created failed")
cg,
[])
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for each_vol in model_update[1]:
self.assertEqual('deleted', each_vol['status'])
cgsnapshot,
snapshots,
None, None))
- self.assertEqual('available',
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE,
model_update['status'],
"CG create from src created failed")
model_update = self.driver.delete_consistencygroup(self.ctxt,
cg, [])
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for each_vol in model_update[1]:
self.assertEqual('deleted', each_vol['status'])
cgsnapshot,
[])
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for volume in model_update[1]:
self.assertEqual('deleted', volume['status'])
source_cg,
[])
- self.assertEqual('deleted', model_update[0]['status'])
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED,
+ model_update[0]['status'])
for each_vol in model_update[1]:
self.assertEqual('deleted', each_vol['status'])
@mock.patch.object(CGQUOTAS, "rollback")
@mock.patch.object(driver.VolumeDriver,
"delete_consistencygroup",
- return_value=({'status': 'deleted'}, []))
+ return_value=({'status': (
+ fields.ConsistencyGroupStatus.DELETED)}, []))
def test_create_delete_consistencygroup(self, fake_delete_cg,
fake_rollback,
fake_commit, fake_reserve):
msg = self.notifier.notifications[0]
self.assertEqual('consistencygroup.create.start', msg['event_type'])
expected = {
- 'status': 'available',
+ 'status': fields.ConsistencyGroupStatus.AVAILABLE,
'name': 'test_cg',
'availability_zone': 'nova',
'tenant_id': self.context.project_id,
self.volume.delete_consistencygroup(self.context, group)
cg = objects.ConsistencyGroup.get_by_id(
context.get_admin_context(read_deleted='yes'), group.id)
- self.assertEqual('deleted', cg.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED, cg.status)
self.assertEqual(4, len(self.notifier.notifications),
self.notifier.notifications)
msg = self.notifier.notifications[2]
self.assertDictMatch(expected, msg['payload'])
msg = self.notifier.notifications[3]
self.assertEqual('consistencygroup.delete.end', msg['event_type'])
- expected['status'] = 'deleted'
+ expected['status'] = fields.ConsistencyGroupStatus.DELETED
self.assertDictMatch(expected, msg['payload'])
self.assertRaises(exception.NotFound,
objects.ConsistencyGroup.get_by_id,
self.volume.create_volume(self.context, volume_id2)
fake_update_cg.return_value = (
- {'status': 'available'},
+ {'status': fields.ConsistencyGroupStatus.AVAILABLE},
[{'id': volume_id2, 'status': 'available'}],
[{'id': volume_id, 'status': 'available'}])
remove_volumes=volume_id)
cg = objects.ConsistencyGroup.get_by_id(self.context, group.id)
expected = {
- 'status': 'available',
+ 'status': fields.ConsistencyGroupStatus.AVAILABLE,
'name': 'test_cg',
'availability_zone': 'nova',
'tenant_id': self.context.project_id,
'user_id': 'fake',
'consistencygroup_id': group.id
}
- self.assertEqual('available', cg.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE, cg.status)
self.assertEqual(10, len(self.notifier.notifications),
self.notifier.notifications)
msg = self.notifier.notifications[6]
self.context,
availability_zone=CONF.storage_availability_zone,
volume_type='type1,type2',
- status='available')
+ status=fields.ConsistencyGroupStatus.AVAILABLE)
volume = tests_utils.create_volume(
self.context,
consistencygroup_id=group.id,
self.context, group2, cgsnapshot=cgsnapshot)
cg2 = objects.ConsistencyGroup.get_by_id(self.context, group2.id)
expected = {
- 'status': 'available',
+ 'status': fields.ConsistencyGroupStatus.AVAILABLE,
'name': 'test_cg',
'availability_zone': 'nova',
'tenant_id': self.context.project_id,
'user_id': 'fake',
'consistencygroup_id': group2.id,
}
- self.assertEqual('available', cg2.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE, cg2.status)
self.assertEqual(group2.id, cg2['id'])
self.assertEqual(cgsnapshot.id, cg2['cgsnapshot_id'])
self.assertIsNone(cg2['source_cgid'])
msg = self.notifier.notifications[6]
self.assertEqual('consistencygroup.delete.start', msg['event_type'])
- expected['status'] = 'available'
+ expected['status'] = fields.ConsistencyGroupStatus.AVAILABLE
self.assertDictMatch(expected, msg['payload'])
msg = self.notifier.notifications[8]
self.assertEqual('consistencygroup.delete.end', msg['event_type'])
- expected['status'] = 'deleted'
+ expected['status'] = fields.ConsistencyGroupStatus.DELETED
self.assertDictMatch(expected, msg['payload'])
cg2 = objects.ConsistencyGroup.get_by_id(
context.get_admin_context(read_deleted='yes'), group2.id)
- self.assertEqual('deleted', cg2.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED, cg2.status)
self.assertRaises(exception.NotFound,
objects.ConsistencyGroup.get_by_id,
self.context,
cg3 = objects.ConsistencyGroup.get_by_id(self.context, group3.id)
- self.assertEqual('available', cg3.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE, cg3.status)
self.assertEqual(group3.id, cg3.id)
self.assertEqual(group.id, cg3.source_cgid)
self.assertIsNone(cg3.cgsnapshot_id)
cg = objects.ConsistencyGroup.get_by_id(
context.get_admin_context(read_deleted='yes'),
group.id)
- self.assertEqual('deleted', cg.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.DELETED, cg.status)
self.assertRaises(exception.NotFound,
objects.ConsistencyGroup.get_by_id,
self.context,
group)
cg = objects.ConsistencyGroup.get_by_id(self.context, group.id)
# Group is not deleted
- self.assertEqual('available', cg.status)
+ self.assertEqual(fields.ConsistencyGroupStatus.AVAILABLE, cg.status)
def test_create_volume_with_consistencygroup_invalid_type(self):
"""Test volume creation with ConsistencyGroup & invalid volume type."""
host='test_host@fakedrv#fakepool',
name='test_cg',
description='this is a test cg',
- status='available',
+ status=fields.ConsistencyGroupStatus.AVAILABLE,
availability_zone='fake_az',
volume_type_id=None,
cgsnapshot_id=None,
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
+from cinder.objects import fields
from cinder.volume.drivers.emc import emc_vmax_fast
from cinder.volume.drivers.emc import emc_vmax_https
from cinder.volume.drivers.emc import emc_vmax_masking
LOG.info(_LI("Create Consistency Group: %(group)s."),
{'group': group['id']})
- modelUpdate = {'status': 'available'}
+ modelUpdate = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
volumeTypeId = group['volume_type_id'].replace(",", "")
cgName = self.utils.truncate_string(group['id'], 8)
"This adds and/or removes volumes from a CG."),
{'group': group['id']})
- modelUpdate = {'status': 'available'}
+ modelUpdate = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
volumeTypeId = group['volume_type_id'].replace(",", "")
cg_name = self.utils.truncate_string(group['id'], 8)
raise exception.VolumeBackendAPIException(
data=exceptionMessage)
- modelUpdate = {'status': 'available'}
+ modelUpdate = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
_poolInstanceName, storageSystem = (
self._get_pool_and_storage_system(extraSpecs))
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
+from cinder.objects import fields
from cinder import utils
from cinder.volume import configuration as config
from cinder.volume.drivers.san import san
self._consistencygroup_creation_check(group)
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
try:
self._client.create_consistencygroup(group['id'])
except Exception:
add_volumes,
remove_volumes):
"""Adds or removes LUN(s) to/from an existing consistency group"""
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
cg_name = group['id']
add_ids = [six.text_type(self.get_lun_id(vol))
for vol in add_volumes] if add_volumes else []
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
from cinder import objects
+from cinder.objects import fields
from cinder import utils
from cinder.volume import driver
from cinder.volume.drivers.san import san
create_data = {'consistency-group-name': group['id']}
self.client.req('consistency-groups', 'POST', data=create_data,
ver='v2')
- return {'status': 'available'}
+ return {'status': fields.ConsistencyGroupStatus.AVAILABLE}
def delete_consistencygroup(self, context, group, volumes):
"""Deletes a consistency group."""
from cinder import exception
from cinder import flow_utils
from cinder.i18n import _, _LE, _LI, _LW
+from cinder.objects import fields
from cinder.volume import qos_specs
from cinder.volume import utils as volume_utils
from cinder.volume import volume_types
self.client.createVolumeSet(cg_name, domain=domain,
comment=six.text_type(extra))
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
return model_update
def create_consistencygroup_from_src(self, context, group, volumes,
from cinder import context
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
+from cinder.objects import fields
from cinder.volume import driver
from cinder.volume import utils
from cinder.volume import volume_types
def create_consistencygroup(self, context, group):
"""Creates a consistencygroup."""
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
return model_update
def create_consistencygroup_from_src(self, context, group, volumes,
from cinder import exception
from cinder.i18n import _, _LE, _LI
from cinder.image import image_utils
+from cinder.objects import fields
from cinder import utils
from cinder.volume import driver
from cinder.volume.drivers import nfs
LOG.error(msg)
raise exception.VolumeBackendAPIException(data=msg)
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
return model_update
def delete_consistencygroup(self, context, group, volumes):
from cinder import context
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
+from cinder.objects import fields
from cinder.volume import driver
from cinder.volume.drivers.ibm.storwize_svc import (
replication as storwize_rep)
LOG.debug('Enter: create_cg_from_source: cg %(cg)s'
' source %(source)s, target %(target)s',
{'cg': fc_consistgrp, 'source': sources, 'target': targets})
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
ctxt = context.get_admin_context()
try:
for source, target in zip(sources, targets):
volumes_model_update = self._get_volume_model_updates(
ctxt, targets, group['id'], model_update['status'])
except exception.VolumeBackendAPIException as err:
- model_update['status'] = 'error'
+ model_update['status'] = fields.ConsistencyGroupStatus.ERROR
volumes_model_update = self._get_volume_model_updates(
ctxt, targets, group['id'], model_update['status'])
with excutils.save_and_reraise_exception():
db will maintain the volumes and CG relationship.
"""
LOG.debug("Creating consistency group.")
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
return model_update
def delete_consistencygroup(self, context, group, volumes):
"""
LOG.debug("Deleting consistency group.")
model_update = {}
- model_update['status'] = 'deleted'
+ model_update['status'] = fields.ConsistencyGroupStatus.DELETED
volumes = self.db.volume_get_all_by_group(context, group['id'])
for volume in volumes:
from cinder import exception
from cinder.i18n import _, _LI, _LW, _LE
from cinder import objects
+from cinder.objects import fields
from cinder.volume import driver
from cinder.volume.drivers.prophetstor import options
from cinder.volume.drivers.san import san
LOG.info(_LI('Start to create consistency group: %(group_name)s '
'id: %(id)s'),
{'group_name': group['name'], 'id': group['id']})
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
try:
ret, output = self.dpl.create_vg(
self._conver_uuid2hex(group['id']),
except Exception:
ret = errno.EFAULT
volume_ref['status'] = 'error_deleting'
- model_update['status'] = 'error_deleting'
+ model_update['status'] = (
+ fields.ConsistencyGroupStatus.ERROR_DELETING)
if ret == 0:
- model_update['status'] = 'deleted'
+ model_update['status'] = fields.ConsistencyGroupStatus.DELETED
return model_update, volumes
def create_cgsnapshot(self, context, cgsnapshot, snapshots):
removevollist = []
cgid = group['id']
vid = ''
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
# Get current group info in backend storage.
ret, output = self.dpl.get_vg(self._conver_uuid2hex(cgid))
if ret == 0:
from cinder import exception
from cinder.i18n import _, _LE, _LI, _LW
from cinder import objects
+from cinder.objects import fields
from cinder import utils
from cinder.volume import driver
from cinder.volume.drivers.san import san
self._array.create_pgroup(self._get_pgroup_name_from_id(group.id))
- model_update = {'status': 'available'}
+ model_update = {'status': fields.ConsistencyGroupStatus.AVAILABLE}
return model_update
def _create_cg_from_cgsnap(self, volumes, snapshots):
from cinder import exception
from cinder.i18n import _, _LE, _LW
from cinder.image import image_utils
+from cinder.objects import fields
from cinder.volume.drivers.san import san
from cinder.volume import qos_specs
from cinder.volume.targets import iscsi as iscsi_driver
# volume associations. So, we're just going to play along with the
# consistency group song and dance. There will be a lot of no-ops
# because of this.
- return {'status': 'available'}
+ return {'status': fields.ConsistencyGroupStatus.AVAILABLE}
def create_consistencygroup_from_src(self, ctxt, group, volumes,
cgsnapshot, snapshots,
snap['id'],
sf_group_snap,
vol))
- return {'status': 'available'}, vol_models
+ return ({'status': fields.ConsistencyGroupStatus.AVAILABLE},
+ vol_models)
elif source_cg and source_vols:
# Create temporary group snapshot.
from cinder.image import glance
from cinder import manager
from cinder import objects
+from cinder.objects import fields
from cinder import quota
from cinder import utils
from cinder import volume as cinder_volume
"""Creates the consistency group."""
context = context.elevated()
- status = 'available'
+ status = fields.ConsistencyGroupStatus.AVAILABLE
model_update = None
self._notify_about_consistencygroup_usage(
group)
if model_update:
- if model_update['status'] == 'error':
+ if (model_update['status'] ==
+ fields.ConsistencyGroupStatus.ERROR):
msg = (_('Create consistency group failed.'))
LOG.error(msg,
resource={'type': 'consistency_group',
group.save()
except Exception:
with excutils.save_and_reraise_exception():
- group.status = 'error'
+ group.status = fields.ConsistencyGroupStatus.ERROR
group.save()
LOG.error(_LE("Consistency group %s: create failed"),
group.name)
self.db.volume_update(context, update['id'], update)
if model_update:
- if model_update['status'] in ['error']:
+ if model_update['status'] in (
+ [fields.ConsistencyGroupStatus.ERROR]):
msg = (_('Error occurred when updating consistency group '
'%s.') % group.id)
LOG.error(msg)
Code Review / openstack-build / cinder-build.git / commitdiff