663 lines
32 KiB
Python
663 lines
32 KiB
Python
# Copyright 2015 SimpliVity Corp.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
import ddt
|
|
import mock
|
|
from oslo_utils import timeutils
|
|
import pytz
|
|
import six
|
|
|
|
from cinder import context
|
|
from cinder import exception
|
|
from cinder import objects
|
|
from cinder.objects import base as ovo_base
|
|
from cinder.objects import fields
|
|
from cinder.tests.unit.consistencygroup import fake_consistencygroup
|
|
from cinder.tests.unit import fake_constants as fake
|
|
from cinder.tests.unit import fake_snapshot
|
|
from cinder.tests.unit import fake_volume
|
|
from cinder.tests.unit import objects as test_objects
|
|
|
|
|
|
@ddt.ddt
|
|
class TestVolume(test_objects.BaseObjectsTestCase):
|
|
@staticmethod
|
|
def _compare(test, db, obj):
|
|
db = {k: v for k, v in db.items()
|
|
if not k.endswith('metadata') or k.startswith('volume')}
|
|
test_objects.BaseObjectsTestCase._compare(test, db, obj)
|
|
|
|
@mock.patch('cinder.db.sqlalchemy.api.volume_get')
|
|
def test_get_by_id(self, volume_get):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume_get.return_value = db_volume
|
|
volume = objects.Volume.get_by_id(self.context, fake.VOLUME_ID)
|
|
volume_get.assert_called_once_with(self.context, fake.VOLUME_ID)
|
|
self._compare(self, db_volume, volume)
|
|
|
|
@mock.patch('cinder.db.sqlalchemy.api.model_query')
|
|
def test_get_by_id_no_existing_id(self, model_query):
|
|
pf = (model_query().options().options().options().options().options().
|
|
options())
|
|
pf.filter_by().first.return_value = None
|
|
self.assertRaises(exception.VolumeNotFound,
|
|
objects.Volume.get_by_id, self.context, 123)
|
|
|
|
@mock.patch('cinder.db.volume_create')
|
|
def test_create(self, volume_create):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume_create.return_value = db_volume
|
|
volume = objects.Volume(context=self.context)
|
|
volume.create()
|
|
self.assertEqual(db_volume['id'], volume.id)
|
|
|
|
@mock.patch('cinder.db.volume_update')
|
|
@ddt.data(False, True)
|
|
def test_save(self, test_cg, volume_update):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.display_name = 'foobar'
|
|
if test_cg:
|
|
volume.consistencygroup = None
|
|
volume.save()
|
|
volume_update.assert_called_once_with(self.context, volume.id,
|
|
{'display_name': 'foobar'})
|
|
|
|
def test_save_error(self):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.display_name = 'foobar'
|
|
volume.consistencygroup = (
|
|
fake_consistencygroup.fake_consistencyobject_obj(self.context))
|
|
self.assertRaises(exception.ObjectActionError,
|
|
volume.save)
|
|
|
|
@mock.patch('cinder.db.volume_metadata_update',
|
|
return_value={'key1': 'value1'})
|
|
@mock.patch('cinder.db.volume_update')
|
|
def test_save_with_metadata(self, volume_update, metadata_update):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.display_name = 'foobar'
|
|
volume.metadata = {'key1': 'value1'}
|
|
self.assertEqual({'display_name': 'foobar',
|
|
'metadata': {'key1': 'value1'}},
|
|
volume.obj_get_changes())
|
|
volume.save()
|
|
volume_update.assert_called_once_with(self.context, volume.id,
|
|
{'display_name': 'foobar'})
|
|
metadata_update.assert_called_once_with(self.context, volume.id,
|
|
{'key1': 'value1'}, True)
|
|
|
|
@mock.patch('cinder.db.volume_admin_metadata_update',
|
|
return_value={'key1': 'value1'})
|
|
@mock.patch('cinder.db.volume_update')
|
|
def test_save_with_admin_metadata(self, volume_update,
|
|
admin_metadata_update):
|
|
# Test with no admin context
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.admin_metadata = {'key1': 'value1'}
|
|
volume.save()
|
|
self.assertFalse(admin_metadata_update.called)
|
|
|
|
# Test with admin context
|
|
admin_context = context.RequestContext(self.user_id, self.project_id,
|
|
is_admin=True)
|
|
volume = objects.Volume._from_db_object(admin_context,
|
|
objects.Volume(), db_volume)
|
|
volume.admin_metadata = {'key1': 'value1'}
|
|
volume.save()
|
|
admin_metadata_update.assert_called_once_with(
|
|
admin_context, volume.id, {'key1': 'value1'}, True)
|
|
|
|
def test_save_with_glance_metadata(self):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.display_name = 'foobar'
|
|
volume.glance_metadata = {'key1': 'value1'}
|
|
self.assertRaises(exception.ObjectActionError, volume.save)
|
|
|
|
def test_save_with_consistencygroup(self):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.display_name = 'foobar'
|
|
volume.consistencygroup = objects.ConsistencyGroup()
|
|
self.assertRaises(exception.ObjectActionError, volume.save)
|
|
|
|
def test_save_with_snapshots(self):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.display_name = 'foobar'
|
|
volume.snapshots = objects.SnapshotList()
|
|
self.assertRaises(exception.ObjectActionError, volume.save)
|
|
|
|
@mock.patch('oslo_utils.timeutils.utcnow', return_value=timeutils.utcnow())
|
|
@mock.patch('cinder.db.sqlalchemy.api.volume_destroy')
|
|
def test_destroy(self, volume_destroy, utcnow_mock):
|
|
volume_destroy.return_value = {
|
|
'status': 'deleted',
|
|
'deleted': True,
|
|
'deleted_at': utcnow_mock.return_value}
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume = objects.Volume._from_db_object(self.context,
|
|
objects.Volume(), db_volume)
|
|
volume.destroy()
|
|
self.assertTrue(volume_destroy.called)
|
|
admin_context = volume_destroy.call_args[0][0]
|
|
self.assertTrue(admin_context.is_admin)
|
|
self.assertTrue(volume.deleted)
|
|
self.assertEqual('deleted', volume.status)
|
|
self.assertEqual(utcnow_mock.return_value.replace(tzinfo=pytz.UTC),
|
|
volume.deleted_at)
|
|
self.assertIsNone(volume.migration_status)
|
|
|
|
def test_obj_fields(self):
|
|
volume = objects.Volume(context=self.context, id=fake.VOLUME_ID,
|
|
name_id=fake.VOLUME_NAME_ID)
|
|
self.assertEqual(['name', 'name_id', 'volume_metadata',
|
|
'volume_admin_metadata', 'volume_glance_metadata'],
|
|
volume.obj_extra_fields)
|
|
self.assertEqual('volume-%s' % fake.VOLUME_NAME_ID, volume.name)
|
|
self.assertEqual(fake.VOLUME_NAME_ID, volume.name_id)
|
|
|
|
def test_obj_field_previous_status(self):
|
|
volume = objects.Volume(context=self.context,
|
|
previous_status='backing-up')
|
|
self.assertEqual('backing-up', volume.previous_status)
|
|
|
|
@mock.patch('cinder.db.volume_metadata_delete')
|
|
def test_delete_metadata_key(self, metadata_delete):
|
|
volume = objects.Volume(self.context, id=fake.VOLUME_ID)
|
|
volume.metadata = {'key1': 'value1', 'key2': 'value2'}
|
|
self.assertEqual({}, volume._orig_metadata)
|
|
volume.delete_metadata_key('key2')
|
|
self.assertEqual({'key1': 'value1'}, volume.metadata)
|
|
metadata_delete.assert_called_once_with(self.context, fake.VOLUME_ID,
|
|
'key2')
|
|
|
|
@mock.patch('cinder.db.volume_metadata_get')
|
|
@mock.patch('cinder.db.volume_glance_metadata_get')
|
|
@mock.patch('cinder.db.volume_admin_metadata_get')
|
|
@mock.patch('cinder.objects.volume_type.VolumeType.get_by_id')
|
|
@mock.patch('cinder.objects.volume_attachment.VolumeAttachmentList.'
|
|
'get_all_by_volume_id')
|
|
@mock.patch('cinder.objects.consistencygroup.ConsistencyGroup.get_by_id')
|
|
@mock.patch('cinder.objects.snapshot.SnapshotList.get_all_for_volume')
|
|
def test_obj_load_attr(self, mock_sl_get_all_for_volume, mock_cg_get_by_id,
|
|
mock_va_get_all_by_vol, mock_vt_get_by_id,
|
|
mock_admin_metadata_get, mock_glance_metadata_get,
|
|
mock_metadata_get):
|
|
fake_db_volume = fake_volume.fake_db_volume(
|
|
consistencygroup_id=fake.CONSISTENCY_GROUP_ID)
|
|
volume = objects.Volume._from_db_object(
|
|
self.context, objects.Volume(), fake_db_volume)
|
|
|
|
# Test metadata lazy-loaded field
|
|
metadata = {'foo': 'bar'}
|
|
mock_metadata_get.return_value = metadata
|
|
self.assertEqual(metadata, volume.metadata)
|
|
mock_metadata_get.assert_called_once_with(self.context, volume.id)
|
|
|
|
# Test glance_metadata lazy-loaded field
|
|
glance_metadata = [{'key': 'foo', 'value': 'bar'}]
|
|
mock_glance_metadata_get.return_value = glance_metadata
|
|
self.assertEqual({'foo': 'bar'}, volume.glance_metadata)
|
|
mock_glance_metadata_get.assert_called_once_with(
|
|
self.context, volume.id)
|
|
|
|
# Test volume_type lazy-loaded field
|
|
# Case1. volume.volume_type_id = None
|
|
self.assertIsNone(volume.volume_type)
|
|
|
|
# Case2. volume2.volume_type_id = 1
|
|
fake2 = fake_volume.fake_db_volume()
|
|
fake2.update({'volume_type_id': fake.VOLUME_ID})
|
|
volume2 = objects.Volume._from_db_object(
|
|
self.context, objects.Volume(), fake2)
|
|
volume_type = objects.VolumeType(context=self.context,
|
|
id=fake.VOLUME_TYPE_ID)
|
|
mock_vt_get_by_id.return_value = volume_type
|
|
self.assertEqual(volume_type, volume2.volume_type)
|
|
mock_vt_get_by_id.assert_called_once_with(self.context,
|
|
volume2.volume_type_id)
|
|
|
|
# Test consistencygroup lazy-loaded field
|
|
consistencygroup = objects.ConsistencyGroup(
|
|
context=self.context, id=fake.CONSISTENCY_GROUP_ID)
|
|
mock_cg_get_by_id.return_value = consistencygroup
|
|
self.assertEqual(consistencygroup, volume.consistencygroup)
|
|
mock_cg_get_by_id.assert_called_once_with(self.context,
|
|
volume.consistencygroup_id)
|
|
|
|
# Test snapshots lazy-loaded field
|
|
snapshots = objects.SnapshotList(context=self.context,
|
|
id=fake.SNAPSHOT_ID)
|
|
mock_sl_get_all_for_volume.return_value = snapshots
|
|
self.assertEqual(snapshots, volume.snapshots)
|
|
mock_sl_get_all_for_volume.assert_called_once_with(self.context,
|
|
volume.id)
|
|
|
|
# Test volume_attachment lazy-loaded field
|
|
va_objs = [objects.VolumeAttachment(context=self.context, id=i)
|
|
for i in [fake.OBJECT_ID, fake.OBJECT2_ID, fake.OBJECT3_ID]]
|
|
va_list = objects.VolumeAttachmentList(context=self.context,
|
|
objects=va_objs)
|
|
mock_va_get_all_by_vol.return_value = va_list
|
|
self.assertEqual(va_list, volume.volume_attachment)
|
|
mock_va_get_all_by_vol.assert_called_once_with(self.context, volume.id)
|
|
|
|
# Test admin_metadata lazy-loaded field - user context
|
|
adm_metadata = {'bar': 'foo'}
|
|
mock_admin_metadata_get.return_value = adm_metadata
|
|
self.assertEqual({}, volume.admin_metadata)
|
|
self.assertFalse(mock_admin_metadata_get.called)
|
|
|
|
# Test admin_metadata lazy-loaded field - admin context
|
|
adm_context = self.context.elevated()
|
|
volume = objects.Volume._from_db_object(adm_context, objects.Volume(),
|
|
fake_volume.fake_db_volume())
|
|
adm_metadata = {'bar': 'foo'}
|
|
mock_admin_metadata_get.return_value = adm_metadata
|
|
self.assertEqual(adm_metadata, volume.admin_metadata)
|
|
mock_admin_metadata_get.assert_called_once_with(adm_context, volume.id)
|
|
|
|
@mock.patch('cinder.objects.consistencygroup.ConsistencyGroup.get_by_id')
|
|
def test_obj_load_attr_cgroup_not_exist(self, mock_cg_get_by_id):
|
|
fake_db_volume = fake_volume.fake_db_volume(consistencygroup_id=None)
|
|
volume = objects.Volume._from_db_object(
|
|
self.context, objects.Volume(), fake_db_volume)
|
|
|
|
self.assertIsNone(volume.consistencygroup)
|
|
mock_cg_get_by_id.assert_not_called()
|
|
|
|
@mock.patch('cinder.objects.group.Group.get_by_id')
|
|
def test_obj_load_attr_group_not_exist(self, mock_group_get_by_id):
|
|
fake_db_volume = fake_volume.fake_db_volume(group_id=None)
|
|
volume = objects.Volume._from_db_object(
|
|
self.context, objects.Volume(), fake_db_volume)
|
|
|
|
self.assertIsNone(volume.group)
|
|
mock_group_get_by_id.assert_not_called()
|
|
|
|
def test_from_db_object_with_all_expected_attributes(self):
|
|
expected_attrs = ['metadata', 'admin_metadata', 'glance_metadata',
|
|
'volume_type', 'volume_attachment',
|
|
'consistencygroup']
|
|
|
|
db_metadata = [{'key': 'foo', 'value': 'bar'}]
|
|
db_admin_metadata = [{'key': 'admin_foo', 'value': 'admin_bar'}]
|
|
db_glance_metadata = [{'key': 'glance_foo', 'value': 'glance_bar'}]
|
|
db_volume_type = fake_volume.fake_db_volume_type()
|
|
db_volume_attachments = fake_volume.volume_attachment_db_obj()
|
|
db_consistencygroup = fake_consistencygroup.fake_db_consistencygroup()
|
|
db_snapshots = fake_snapshot.fake_db_snapshot()
|
|
|
|
db_volume = fake_volume.fake_db_volume(
|
|
volume_metadata=db_metadata,
|
|
volume_admin_metadata=db_admin_metadata,
|
|
volume_glance_metadata=db_glance_metadata,
|
|
volume_type=db_volume_type,
|
|
volume_attachment=[db_volume_attachments],
|
|
consistencygroup=db_consistencygroup,
|
|
snapshots=[db_snapshots],
|
|
)
|
|
volume = objects.Volume._from_db_object(self.context, objects.Volume(),
|
|
db_volume, expected_attrs)
|
|
|
|
self.assertEqual({'foo': 'bar'}, volume.metadata)
|
|
self.assertEqual({'admin_foo': 'admin_bar'}, volume.admin_metadata)
|
|
self.assertEqual({'glance_foo': 'glance_bar'}, volume.glance_metadata)
|
|
self._compare(self, db_volume_type, volume.volume_type)
|
|
self._compare(self, db_volume_attachments, volume.volume_attachment)
|
|
self._compare(self, db_consistencygroup, volume.consistencygroup)
|
|
self._compare(self, db_snapshots, volume.snapshots)
|
|
|
|
@mock.patch('cinder.db.volume_glance_metadata_get', return_value={})
|
|
@mock.patch('cinder.db.sqlalchemy.api.volume_get')
|
|
def test_refresh(self, volume_get, volume_metadata_get):
|
|
db_volume1 = fake_volume.fake_db_volume()
|
|
db_volume2 = db_volume1.copy()
|
|
db_volume2['display_name'] = 'foobar'
|
|
|
|
# On the second volume_get, return the volume with an updated
|
|
# display_name
|
|
volume_get.side_effect = [db_volume1, db_volume2]
|
|
volume = objects.Volume.get_by_id(self.context, fake.VOLUME_ID)
|
|
self._compare(self, db_volume1, volume)
|
|
|
|
# display_name was updated, so a volume refresh should have a new value
|
|
# for that field
|
|
volume.refresh()
|
|
self._compare(self, db_volume2, volume)
|
|
if six.PY3:
|
|
call_bool = mock.call.__bool__()
|
|
else:
|
|
call_bool = mock.call.__nonzero__()
|
|
volume_get.assert_has_calls([mock.call(self.context, fake.VOLUME_ID),
|
|
call_bool,
|
|
mock.call(self.context, fake.VOLUME_ID)])
|
|
|
|
def test_metadata_aliases(self):
|
|
volume = objects.Volume(context=self.context)
|
|
# metadata<->volume_metadata
|
|
volume.metadata = {'abc': 'def'}
|
|
self.assertEqual([{'key': 'abc', 'value': 'def'}],
|
|
volume.volume_metadata)
|
|
|
|
md = [{'key': 'def', 'value': 'abc'}]
|
|
volume.volume_metadata = md
|
|
self.assertEqual({'def': 'abc'}, volume.metadata)
|
|
|
|
# admin_metadata<->volume_admin_metadata
|
|
volume.admin_metadata = {'foo': 'bar'}
|
|
self.assertEqual([{'key': 'foo', 'value': 'bar'}],
|
|
volume.volume_admin_metadata)
|
|
|
|
volume.volume_admin_metadata = [{'key': 'xyz', 'value': '42'}]
|
|
self.assertEqual({'xyz': '42'}, volume.admin_metadata)
|
|
|
|
# glance_metadata<->volume_glance_metadata
|
|
volume.glance_metadata = {'jkl': 'mno'}
|
|
self.assertEqual([{'key': 'jkl', 'value': 'mno'}],
|
|
volume.volume_glance_metadata)
|
|
|
|
volume.volume_glance_metadata = [{'key': 'prs', 'value': 'tuw'}]
|
|
self.assertEqual({'prs': 'tuw'}, volume.glance_metadata)
|
|
|
|
@mock.patch('cinder.db.volume_metadata_update', return_value={})
|
|
@mock.patch('cinder.db.volume_update')
|
|
@ddt.data({'src_vol_type_id': fake.VOLUME_TYPE_ID,
|
|
'dest_vol_type_id': fake.VOLUME_TYPE2_ID},
|
|
{'src_vol_type_id': None,
|
|
'dest_vol_type_id': fake.VOLUME_TYPE2_ID})
|
|
@ddt.unpack
|
|
def test_finish_volume_migration(self, volume_update, metadata_update,
|
|
src_vol_type_id, dest_vol_type_id):
|
|
src_volume_db = fake_volume.fake_db_volume(
|
|
**{'id': fake.VOLUME_ID, 'volume_type_id': src_vol_type_id})
|
|
if src_vol_type_id:
|
|
src_volume_db['volume_type'] = fake_volume.fake_db_volume_type(
|
|
id=src_vol_type_id)
|
|
dest_volume_db = fake_volume.fake_db_volume(
|
|
**{'id': fake.VOLUME2_ID, 'volume_type_id': dest_vol_type_id})
|
|
if dest_vol_type_id:
|
|
dest_volume_db['volume_type'] = fake_volume.fake_db_volume_type(
|
|
id=dest_vol_type_id)
|
|
expected_attrs = objects.Volume._get_expected_attrs(self.context)
|
|
src_volume = objects.Volume._from_db_object(
|
|
self.context, objects.Volume(), src_volume_db,
|
|
expected_attrs=expected_attrs)
|
|
dest_volume = objects.Volume._from_db_object(
|
|
self.context, objects.Volume(), dest_volume_db,
|
|
expected_attrs=expected_attrs)
|
|
updated_dest_volume = src_volume.finish_volume_migration(
|
|
dest_volume)
|
|
self.assertEqual('deleting', updated_dest_volume.migration_status)
|
|
self.assertEqual('migration src for ' + src_volume.id,
|
|
updated_dest_volume.display_description)
|
|
self.assertEqual(src_volume.id, updated_dest_volume._name_id)
|
|
self.assertTrue(volume_update.called)
|
|
volume_update.assert_has_calls([
|
|
mock.call(self.context, src_volume.id, mock.ANY),
|
|
mock.call(self.context, dest_volume.id, mock.ANY)])
|
|
ctxt, vol_id, updates = volume_update.call_args[0]
|
|
self.assertNotIn('volume_type', updates)
|
|
|
|
# Ensure that the destination volume type has not been overwritten
|
|
self.assertEqual(dest_vol_type_id,
|
|
getattr(updated_dest_volume, 'volume_type_id'))
|
|
# Ignore these attributes, since they were updated by
|
|
# finish_volume_migration
|
|
ignore_keys = ('id', 'provider_location', '_name_id',
|
|
'migration_status', 'display_description', 'status',
|
|
'volume_glance_metadata', 'volume_type')
|
|
|
|
dest_vol_dict = {k: updated_dest_volume[k] for k in
|
|
updated_dest_volume.keys() if k not in ignore_keys}
|
|
src_vol_dict = {k: src_volume[k] for k in src_volume.keys()
|
|
if k not in ignore_keys}
|
|
self.assertEqual(src_vol_dict, dest_vol_dict)
|
|
|
|
def test_volume_with_metadata_serialize_deserialize_no_changes(self):
|
|
updates = {'volume_glance_metadata': [{'key': 'foo', 'value': 'bar'}],
|
|
'expected_attrs': ['glance_metadata']}
|
|
volume = fake_volume.fake_volume_obj(self.context, **updates)
|
|
serializer = objects.base.CinderObjectSerializer()
|
|
serialized_volume = serializer.serialize_entity(self.context, volume)
|
|
volume = serializer.deserialize_entity(self.context, serialized_volume)
|
|
self.assertDictEqual({}, volume.obj_get_changes())
|
|
|
|
@mock.patch('cinder.db.volume_admin_metadata_update')
|
|
@mock.patch('cinder.db.sqlalchemy.api.volume_attach')
|
|
def test_begin_attach(self, volume_attach, metadata_update):
|
|
volume = fake_volume.fake_volume_obj(self.context)
|
|
db_attachment = fake_volume.volume_attachment_db_obj(
|
|
volume_id=volume.id,
|
|
attach_status=fields.VolumeAttachStatus.ATTACHING)
|
|
volume_attach.return_value = db_attachment
|
|
metadata_update.return_value = {'attached_mode': 'rw'}
|
|
|
|
with mock.patch.object(self.context, 'elevated') as mock_elevated:
|
|
mock_elevated.return_value = context.get_admin_context()
|
|
attachment = volume.begin_attach("rw")
|
|
self.assertIsInstance(attachment, objects.VolumeAttachment)
|
|
self.assertEqual(volume.id, attachment.volume_id)
|
|
self.assertEqual(fields.VolumeAttachStatus.ATTACHING,
|
|
attachment.attach_status)
|
|
metadata_update.assert_called_once_with(self.context.elevated(),
|
|
volume.id,
|
|
{'attached_mode': u'rw'},
|
|
True)
|
|
self.assertEqual('rw', volume.admin_metadata['attached_mode'])
|
|
|
|
@mock.patch('cinder.db.volume_admin_metadata_delete')
|
|
@mock.patch('cinder.db.sqlalchemy.api.volume_detached')
|
|
@mock.patch('cinder.objects.volume_attachment.VolumeAttachmentList.'
|
|
'get_all_by_volume_id')
|
|
def test_volume_detached_with_attachment(
|
|
self, volume_attachment_get,
|
|
volume_detached,
|
|
metadata_delete):
|
|
va_objs = [objects.VolumeAttachment(context=self.context, id=i)
|
|
for i in [fake.OBJECT_ID, fake.OBJECT2_ID, fake.OBJECT3_ID]]
|
|
# As changes are not saved, we need reset it here. Later changes
|
|
# will be checked.
|
|
for obj in va_objs:
|
|
obj.obj_reset_changes()
|
|
va_list = objects.VolumeAttachmentList(context=self.context,
|
|
objects=va_objs)
|
|
va_list.obj_reset_changes()
|
|
volume_attachment_get.return_value = va_list
|
|
admin_context = context.get_admin_context()
|
|
volume = fake_volume.fake_volume_obj(
|
|
admin_context,
|
|
volume_attachment=va_list,
|
|
volume_admin_metadata=[{'key': 'attached_mode',
|
|
'value': 'rw'}])
|
|
self.assertEqual(3, len(volume.volume_attachment))
|
|
volume_detached.return_value = ({'status': 'in-use'},
|
|
{'attached_mode': 'rw'})
|
|
with mock.patch.object(admin_context, 'elevated') as mock_elevated:
|
|
mock_elevated.return_value = admin_context
|
|
volume.finish_detach(fake.OBJECT_ID)
|
|
volume_detached.assert_called_once_with(admin_context,
|
|
volume.id,
|
|
fake.OBJECT_ID)
|
|
metadata_delete.assert_called_once_with(admin_context,
|
|
volume.id,
|
|
'attached_mode')
|
|
self.assertEqual('in-use', volume.status)
|
|
self.assertEqual({}, volume.cinder_obj_get_changes())
|
|
self.assertEqual(2, len(volume.volume_attachment))
|
|
self.assertNotIn('attached_mode', volume.admin_metadata)
|
|
|
|
@mock.patch('cinder.db.volume_admin_metadata_delete')
|
|
@mock.patch('cinder.db.sqlalchemy.api.volume_detached')
|
|
@mock.patch('cinder.objects.volume_attachment.VolumeAttachmentList.'
|
|
'get_all_by_volume_id')
|
|
def test_volume_detached_without_attachment(
|
|
self, volume_attachment_get, volume_detached, metadata_delete):
|
|
admin_context = context.get_admin_context()
|
|
volume = fake_volume.fake_volume_obj(
|
|
admin_context,
|
|
volume_admin_metadata=[{'key': 'attached_mode',
|
|
'value': 'rw'}])
|
|
self.assertFalse(volume.obj_attr_is_set('volume_attachment'))
|
|
volume_detached.return_value = ({'status': 'in-use'}, None)
|
|
with mock.patch.object(admin_context, 'elevated') as mock_elevated:
|
|
mock_elevated.return_value = admin_context
|
|
volume.finish_detach(fake.OBJECT_ID)
|
|
metadata_delete.assert_called_once_with(admin_context,
|
|
volume.id,
|
|
'attached_mode')
|
|
volume_detached.assert_called_once_with(admin_context,
|
|
volume.id,
|
|
fake.OBJECT_ID)
|
|
self.assertEqual('in-use', volume.status)
|
|
self.assertEqual({}, volume.cinder_obj_get_changes())
|
|
self.assertFalse(volume_attachment_get.called)
|
|
|
|
@ddt.data('1.6', '1.7')
|
|
def test_obj_make_compatible_cluster_added(self, version):
|
|
extra_data = {'cluster_name': 'cluster_name',
|
|
'cluster': objects.Cluster()}
|
|
volume = objects.Volume(self.context, host='host', **extra_data)
|
|
|
|
serializer = ovo_base.CinderObjectSerializer(version)
|
|
primitive = serializer.serialize_entity(self.context, volume)
|
|
|
|
converted_volume = objects.Volume.obj_from_primitive(primitive)
|
|
is_set = version == '1.7'
|
|
for key in extra_data:
|
|
self.assertEqual(is_set, converted_volume.obj_attr_is_set(key))
|
|
self.assertEqual('host', converted_volume.host)
|
|
|
|
@ddt.data('1.9', '1.10')
|
|
def test_obj_make_compatible_groups_added(self, version):
|
|
extra_data = {'group_id': fake.GROUP_ID,
|
|
'group': objects.Group()}
|
|
volume = objects.Volume(self.context, host='host', **extra_data)
|
|
|
|
serializer = ovo_base.CinderObjectSerializer(version)
|
|
primitive = serializer.serialize_entity(self.context, volume)
|
|
|
|
converted_volume = objects.Volume.obj_from_primitive(primitive)
|
|
is_set = version == '1.10'
|
|
for key in extra_data:
|
|
self.assertEqual(is_set, converted_volume.obj_attr_is_set(key))
|
|
self.assertEqual('host', converted_volume.host)
|
|
|
|
@ddt.data(True, False)
|
|
def test_is_replicated(self, result):
|
|
volume_type = fake_volume.fake_volume_type_obj(self.context)
|
|
volume = fake_volume.fake_volume_obj(
|
|
self.context, volume_type_id=volume_type.id)
|
|
volume.volume_type = volume_type
|
|
with mock.patch.object(volume_type, 'is_replicated',
|
|
return_value=result) as is_replicated:
|
|
self.assertEqual(result, volume.is_replicated())
|
|
is_replicated.assert_called_once_with()
|
|
|
|
def test_is_replicated_no_type(self):
|
|
volume = fake_volume.fake_volume_obj(
|
|
self.context, volume_type_id=None, volume_type=None)
|
|
self.assertFalse(bool(volume.is_replicated()))
|
|
|
|
|
|
@ddt.ddt
|
|
class TestVolumeList(test_objects.BaseObjectsTestCase):
|
|
@mock.patch('cinder.db.volume_get_all')
|
|
def test_get_all(self, volume_get_all):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
volume_get_all.return_value = [db_volume]
|
|
|
|
volumes = objects.VolumeList.get_all(self.context,
|
|
mock.sentinel.marker,
|
|
mock.sentinel.limit,
|
|
mock.sentinel.sort_key,
|
|
mock.sentinel.sort_dir)
|
|
self.assertEqual(1, len(volumes))
|
|
TestVolume._compare(self, db_volume, volumes[0])
|
|
|
|
@mock.patch('cinder.db.volume_get_all_by_host')
|
|
def test_get_by_host(self, get_all_by_host):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
get_all_by_host.return_value = [db_volume]
|
|
|
|
volumes = objects.VolumeList.get_all_by_host(
|
|
self.context, 'fake-host')
|
|
self.assertEqual(1, len(volumes))
|
|
TestVolume._compare(self, db_volume, volumes[0])
|
|
|
|
@mock.patch('cinder.db.volume_get_all_by_group')
|
|
def test_get_by_group(self, get_all_by_group):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
get_all_by_group.return_value = [db_volume]
|
|
|
|
volumes = objects.VolumeList.get_all_by_group(
|
|
self.context, 'fake-host')
|
|
self.assertEqual(1, len(volumes))
|
|
TestVolume._compare(self, db_volume, volumes[0])
|
|
|
|
@mock.patch('cinder.db.volume_get_all_by_project')
|
|
def test_get_by_project(self, get_all_by_project):
|
|
db_volume = fake_volume.fake_db_volume()
|
|
get_all_by_project.return_value = [db_volume]
|
|
|
|
volumes = objects.VolumeList.get_all_by_project(
|
|
self.context, mock.sentinel.project_id, mock.sentinel.marker,
|
|
mock.sentinel.limit, mock.sentinel.sorted_keys,
|
|
mock.sentinel.sorted_dirs, mock.sentinel.filters)
|
|
self.assertEqual(1, len(volumes))
|
|
TestVolume._compare(self, db_volume, volumes[0])
|
|
|
|
@ddt.data(['name_id'], ['__contains__'])
|
|
def test_get_by_project_with_sort_key(self, sort_keys):
|
|
fake_volume.fake_db_volume()
|
|
|
|
self.assertRaises(exception.InvalidInput,
|
|
objects.VolumeList.get_all_by_project,
|
|
self.context,
|
|
self.context.project_id,
|
|
sort_keys=sort_keys)
|
|
|
|
@mock.patch('cinder.db.volume_include_in_cluster')
|
|
def test_include_in_cluster(self, include_mock):
|
|
filters = {'host': mock.sentinel.host,
|
|
'cluster_name': mock.sentinel.cluster_name}
|
|
cluster = 'new_cluster'
|
|
objects.VolumeList.include_in_cluster(self.context, cluster, **filters)
|
|
include_mock.assert_called_once_with(self.context, cluster, True,
|
|
**filters)
|
|
|
|
@mock.patch('cinder.db.volume_include_in_cluster')
|
|
def test_include_in_cluster_specify_partial(self, include_mock):
|
|
filters = {'host': mock.sentinel.host,
|
|
'cluster_name': mock.sentinel.cluster_name}
|
|
cluster = 'new_cluster'
|
|
objects.VolumeList.include_in_cluster(self.context, cluster,
|
|
mock.sentinel.partial_rename,
|
|
**filters)
|
|
include_mock.assert_called_once_with(self.context, cluster,
|
|
mock.sentinel.partial_rename,
|
|
**filters)
|