def test_add_access(self):
project_id = '456'
- vtype = volume_types.create(self.ctxt, 'type1')
+ vtype = volume_types.create(self.ctxt, 'type1', is_public=False)
vtype_id = vtype.get('id')
volume_types.add_volume_type_access(self.ctxt, vtype_id, project_id)
def test_remove_access(self):
project_id = '456'
- vtype = volume_types.create(self.ctxt, 'type1', projects=['456'])
+ vtype = volume_types.create(self.ctxt, 'type1', projects=['456'],
+ is_public=False)
vtype_id = vtype.get('id')
volume_types.remove_volume_type_access(self.ctxt, vtype_id, project_id)
def test_get_volume_type_encryption_without_volume_type_id(self):
ret = volume_types.get_volume_type_encryption(self.ctxt, None)
self.assertIsNone(ret)
+
+ def test_check_public_volume_type_failed(self):
+ project_id = '456'
+ volume_type = volume_types.create(self.ctxt, "type1")
+ volume_type_id = volume_type.get('id')
+ self.assertRaises(exception.InvalidVolumeType,
+ volume_types.add_volume_type_access,
+ self.ctxt, volume_type_id, project_id)
+ self.assertRaises(exception.InvalidVolumeType,
+ volume_types.remove_volume_type_access,
+ self.ctxt, volume_type_id, project_id)
+
+ def test_check_private_volume_type(self):
+ volume_type = volume_types.create(self.ctxt, "type1", is_public=False)
+ volume_type_id = volume_type.get('id')
+ self.assertFalse(volume_types.is_public_volume_type(self.ctxt,
+ volume_type_id))
return extra_specs
+def is_public_volume_type(context, volume_type_id):
+ """Return is_public boolean value of volume type"""
+ volume_type = db.volume_type_get(context, volume_type_id)
+ return volume_type['is_public']
+
+
def add_volume_type_access(context, volume_type_id, project_id):
"""Add access to volume type for project_id."""
if volume_type_id is None:
msg = _("volume_type_id cannot be None")
raise exception.InvalidVolumeType(reason=msg)
+ if is_public_volume_type(context, volume_type_id):
+ msg = _("Type access modification is not applicable to public volume "
+ "type.")
+ raise exception.InvalidVolumeType(reason=msg)
return db.volume_type_access_add(context, volume_type_id, project_id)
if volume_type_id is None:
msg = _("volume_type_id cannot be None")
raise exception.InvalidVolumeType(reason=msg)
+ if is_public_volume_type(context, volume_type_id):
+ msg = _("Type access modification is not applicable to public volume "
+ "type.")
+ raise exception.InvalidVolumeType(reason=msg)
return db.volume_type_access_remove(context, volume_type_id, project_id)