@require_context
-@require_snapshot_exists
def _snapshot_metadata_get(context, snapshot_id, session=None):
rows = _snapshot_metadata_get_query(context, snapshot_id, session).all()
result = {}
class DotHillNotTargetPortal(CinderException):
message = _("No active iSCSI portals with supplied iSCSI IPs")
+
+
+class MetadataAbsent(CinderException):
+ message = _("There is no metadata in DB object.")
volume._from_db_object(context, volume, db_snapshot['volume'])
snapshot.volume = volume
if 'metadata' in expected_attrs:
- snapshot.metadata = db.snapshot_metadata_get(context,
- db_snapshot['id'])
-
+ metadata = db_snapshot.get('snapshot_metadata')
+ if metadata is None:
+ raise exception.MetadataAbsent()
+ snapshot.metadata = {item['key']: item['value']
+ for item in metadata}
snapshot._context = context
snapshot.obj_reset_changes()
return snapshot
'created_at': None,
'display_name': 'Default name',
'display_description': 'Default description',
- 'project_id': 'fake'}
+ 'project_id': 'fake',
+ 'snapshot_metadata': []}
snapshot.update(kwargs)
return snapshot
CONF = cfg.CONF
-def return_create_snapshot_metadata_max(context,
- snapshot_id,
- metadata,
- delete):
- return stub_max_snapshot_metadata()
-
-
def return_create_snapshot_metadata(context, snapshot_id, metadata, delete):
return stub_snapshot_metadata()
return stub_new_snapshot_metadata()
-def return_snapshot_metadata(context, snapshot_id):
- if not isinstance(snapshot_id, str) or not len(snapshot_id) == 36:
- msg = 'id %s must be a uuid in return snapshot metadata' % snapshot_id
- raise Exception(msg)
- return stub_snapshot_metadata()
-
-
-def return_empty_snapshot_metadata(context, snapshot_id):
- return {}
-
-
def return_empty_container_metadata(context, snapshot_id, metadata, delete):
return {}
-def delete_snapshot_metadata(context, snapshot_id, key):
- pass
-
-
def stub_snapshot_metadata():
metadata = {
"key1": "value1",
return metadata
-def stub_max_snapshot_metadata():
- metadata = {"metadata": {}}
- for num in range(CONF.quota_metadata_items):
- metadata['metadata']['key%i' % num] = "blah"
- return metadata
-
-
def return_snapshot(context, snapshot_id):
return {'id': '0cc3346e-9fef-4445-abe6-5d2b2690ec64',
'name': 'fake',
self.volume_api = cinder.volume.api.API()
self.stubs.Set(cinder.db, 'volume_get', return_volume)
self.stubs.Set(cinder.db, 'snapshot_get', return_snapshot)
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_snapshot_metadata)
self.stubs.Set(self.volume_api, 'update_snapshot_metadata',
fake_update_snapshot_metadata)
req = fakes.HTTPRequest.blank('/v1/snapshots')
self.snapshot_controller.create(req, body)
- @mock.patch('cinder.db.snapshot_metadata_get',
- return_value={'key1': 'value1',
- 'key2': 'value2',
- 'key3': 'value3'})
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_index(self, snapshot_get_by_id, snapshot_metadata_get):
+ def test_index(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext('admin', 'fake', True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
+ snapshot_obj['metadata'] = {'key1': 'value1',
+ 'key2': 'value2',
+ 'key3': 'value3'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url)
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.index, req, self.url)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_index_no_data(self, snapshot_get_by_id, snapshot_metadata_get):
+ def test_index_no_data(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
expected = {'metadata': {}}
self.assertEqual(expected, res_dict)
- @mock.patch('cinder.db.snapshot_metadata_get',
- return_value={'key2': 'value2'})
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_show(self, snapshot_get_by_id, snapshot_metadata_get):
+ def test_show(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext('admin', 'fake', True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
+ snapshot_obj['metadata'] = {'key2': 'value2'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key2')
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, req, self.req_id, 'key2')
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_show_meta_not_found(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_show_meta_not_found(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
self.controller.show, req, self.req_id, 'key6')
@mock.patch('cinder.db.snapshot_metadata_delete')
- @mock.patch('cinder.db.snapshot_metadata_get',
- return_value={'key2': 'value2'})
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_delete(self, snapshot_get_by_id, snapshot_metadata_get,
- snapshot_metadata_delete):
+ def test_delete(self, snapshot_get_by_id, snapshot_metadata_delete):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext('admin', 'fake', True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
+ snapshot_obj['metadata'] = {'key2': 'value2'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key2')
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.delete, req, self.req_id, 'key1')
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_delete_meta_not_found(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_delete_meta_not_found(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
snapshot_get_by_id.return_value = snapshot_obj
volume_get_by_id.return_value = fake_volume_obj
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_empty_snapshot_metadata)
self.stubs.Set(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
res_dict = self.controller.create(req, self.req_id, body)
self.assertEqual(body, res_dict)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.db.snapshot_update')
@mock.patch('cinder.objects.Snapshot.get_by_id')
def test_create_with_keys_in_uppercase_and_lowercase(
- self, snapshot_get_by_id, snapshot_update, snapshot_metadata_get):
+ self, snapshot_get_by_id, snapshot_update):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
def test_create_nonexistent_snapshot(self):
self.stubs.Set(cinder.db, 'snapshot_get',
return_snapshot_nonexistent)
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_snapshot_metadata)
self.stubs.Set(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_create_snapshot_metadata)
self.stubs.Set(cinder.db, 'snapshot_metadata_update',
return_new_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
@mock.patch('cinder.db.snapshot_metadata_update', return_value=dict())
@mock.patch('cinder.db.snapshot_update')
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_update_item(self, snapshot_get_by_id, snapshot_metadata_get,
+ def test_update_item(self, snapshot_get_by_id,
snapshot_update, snapshot_metadata_update):
snapshot = {
'id': self.req_id,
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.update, req, self.req_id, '', body)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_update_item_key_too_long(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_update_item_key_too_long(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
self.controller.update,
req, self.req_id, ("a" * 260), body)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_update_item_value_too_long(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_update_item_value_too_long(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
self.controller.update, req, self.req_id, 'bad',
body)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_invalid_metadata_items_on_create(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_invalid_metadata_items_on_create(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
'created_at': None,
'display_name': 'Default name',
'display_description': 'Default description',
- 'project_id': 'fake'}
+ 'project_id': 'fake',
+ 'snapshot_metadata': []}
snapshot.update(kwargs)
return snapshot
CONF = cfg.CONF
-def return_create_snapshot_metadata_max(context,
- snapshot_id,
- metadata,
- delete):
- return stub_max_snapshot_metadata()
-
-
def return_create_snapshot_metadata(context, snapshot_id, metadata, delete):
return stub_snapshot_metadata()
return stub_new_snapshot_metadata()
-def return_snapshot_metadata(context, snapshot_id):
- if not isinstance(snapshot_id, str) or not len(snapshot_id) == 36:
- msg = 'id %s must be a uuid in return snapshot metadata' % snapshot_id
- raise Exception(msg)
- return stub_snapshot_metadata()
-
-
-def return_empty_snapshot_metadata(context, snapshot_id):
- return {}
-
-
def return_empty_container_metadata(context, snapshot_id, metadata, delete):
return {}
-def delete_snapshot_metadata(context, snapshot_id, key):
- pass
-
-
def stub_snapshot_metadata():
metadata = {
"key1": "value1",
return metadata
-def stub_max_snapshot_metadata():
- metadata = {"metadata": {}}
- for num in range(CONF.quota_metadata_items):
- metadata['metadata']['key%i' % num] = "blah"
- return metadata
-
-
def return_snapshot(context, snapshot_id):
return {'id': '0cc3346e-9fef-4445-abe6-5d2b2690ec64',
'name': 'fake',
self.volume_api = cinder.volume.api.API()
self.stubs.Set(cinder.db, 'volume_get', return_volume)
self.stubs.Set(cinder.db, 'snapshot_get', return_snapshot)
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_snapshot_metadata)
self.stubs.Set(self.volume_api, 'update_snapshot_metadata',
fake_update_snapshot_metadata)
req = fakes.HTTPRequest.blank('/v2/snapshots')
self.snapshot_controller.create(req, body)
- @mock.patch('cinder.db.snapshot_metadata_get',
- return_value={'key1': 'value1',
- 'key2': 'value2',
- 'key3': 'value3'})
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_index(self, snapshot_get_by_id, snapshot_metadata_get):
+ def test_index(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext('admin', 'fake', True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
+ snapshot_obj['metadata'] = {'key1': 'value1',
+ 'key2': 'value2',
+ 'key3': 'value3'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url)
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.index, req, self.url)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_index_no_data(self, snapshot_get_by_id, snapshot_metadata_get):
+ def test_index_no_data(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
expected = {'metadata': {}}
self.assertEqual(expected, res_dict)
- @mock.patch('cinder.db.snapshot_metadata_get',
- return_value={'key2': 'value2'})
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_show(self, snapshot_get_by_id, snapshot_metadata_get):
+ def test_show(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext('admin', 'fake', True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
+ snapshot_obj['metadata'] = {'key2': 'value2'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key2')
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.show, req, self.req_id, 'key2')
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_show_meta_not_found(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_show_meta_not_found(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
self.controller.show, req, self.req_id, 'key6')
@mock.patch('cinder.db.snapshot_metadata_delete')
- @mock.patch('cinder.db.snapshot_metadata_get',
- return_value={'key2': 'value2'})
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_delete(self, snapshot_get_by_id, snapshot_metadata_get,
- snapshot_metadata_delete):
+ def test_delete(self, snapshot_get_by_id, snapshot_metadata_delete):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
}
ctx = context.RequestContext('admin', 'fake', True)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
+ snapshot_obj['metadata'] = {'key2': 'value2'}
snapshot_get_by_id.return_value = snapshot_obj
req = fakes.HTTPRequest.blank(self.url + '/key2')
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.delete, req, self.req_id, 'key1')
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_delete_meta_not_found(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_delete_meta_not_found(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
snapshot_get_by_id.return_value = snapshot_obj
volume_get_by_id.return_value = fake_volume_obj
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_empty_snapshot_metadata)
self.stubs.Set(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
res_dict = self.controller.create(req, self.req_id, body)
self.assertEqual(body, res_dict)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.db.snapshot_update')
@mock.patch('cinder.objects.Snapshot.get_by_id')
def test_create_with_keys_in_uppercase_and_lowercase(
- self, snapshot_get_by_id, snapshot_update, snapshot_metadata_get):
+ self, snapshot_get_by_id, snapshot_update):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
def test_create_nonexistent_snapshot(self):
self.stubs.Set(cinder.db, 'snapshot_get',
return_snapshot_nonexistent)
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_snapshot_metadata)
self.stubs.Set(cinder.db, 'snapshot_metadata_update',
return_create_snapshot_metadata)
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot)
snapshot_get_by_id.return_value = snapshot_obj
- self.stubs.Set(cinder.db, 'snapshot_metadata_get',
- return_create_snapshot_metadata)
self.stubs.Set(cinder.db, 'snapshot_metadata_update',
return_new_snapshot_metadata)
req = fakes.HTTPRequest.blank(self.url)
@mock.patch('cinder.db.snapshot_metadata_update', return_value=dict())
@mock.patch('cinder.db.snapshot_update')
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_update_item(self, snapshot_get_by_id, snapshot_metadata_get,
+ def test_update_item(self, snapshot_get_by_id,
snapshot_update, snapshot_metadata_update):
snapshot = {
'id': self.req_id,
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.update, req, self.req_id, '', body)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_update_item_key_too_long(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_update_item_key_too_long(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
self.controller.update,
req, self.req_id, ("a" * 260), body)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_update_item_value_too_long(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_update_item_value_too_long(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
self.controller.update, req, self.req_id, 'bad',
body)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
@mock.patch('cinder.objects.Snapshot.get_by_id')
- def test_invalid_metadata_items_on_create(self, snapshot_get_by_id,
- snapshot_metadata_get):
+ def test_invalid_metadata_items_on_create(self, snapshot_get_by_id):
snapshot = {
'id': self.req_id,
'expected_attrs': ['metadata']
from oslo_versionedobjects import fields
-from cinder import objects
-
-
-def fake_db_volume(**updates):
- db_volume = {
- 'id': 1,
- 'size': 1,
- 'name': 'fake',
- 'availability_zone': 'fake_availability_zone',
- 'status': 'available',
- 'attach_status': 'detached',
- }
-
- for name, field in objects.Volume.fields.items():
- if name in db_volume:
- continue
- if field.nullable:
- db_volume[name] = None
- elif field.default != fields.UnspecifiedDefault:
- db_volume[name] = field.default
- else:
- raise Exception('fake_db_volume needs help with %s' % name)
-
- if updates:
- db_volume.update(updates)
-
- return db_volume
+from cinder.objects import snapshot
def fake_db_snapshot(**updates):
'display_name': 'fake_name',
'display_description': 'fake_description',
'metadata': {},
- 'snapshot_metadata': {},
+ 'snapshot_metadata': [],
}
- for name, field in objects.Snapshot.fields.items():
+ for name, field in snapshot.Snapshot.fields.items():
if name in db_snapshot:
continue
if field.nullable:
def fake_snapshot_obj(context, **updates):
expected_attrs = updates.pop('expected_attrs', None)
- return objects.Snapshot._from_db_object(context, objects.Snapshot(),
- fake_db_snapshot(**updates),
- expected_attrs=expected_attrs)
-
-
-def fake_volume_obj(context, **updates):
- expected_attrs = updates.pop('expected_attrs', None)
- return objects.Volume._from_db_object(context, objects.Volume(),
- fake_db_volume(**updates),
- expected_attrs=expected_attrs)
+ return snapshot.Snapshot._from_db_object(context, snapshot.Snapshot(),
+ fake_db_snapshot(**updates),
+ expected_attrs=expected_attrs)
# License for the specific language governing permissions and limitations
# under the License.
+import copy
import mock
+from oslo_log import log as logging
+
+from cinder import exception
from cinder import objects
+from cinder.tests.unit import fake_snapshot
from cinder.tests.unit import fake_volume
from cinder.tests.unit import objects as test_objects
-fake_snapshot = {
+
+LOG = logging.getLogger(__name__)
+
+
+fake_db_snapshot = fake_snapshot.fake_db_snapshot()
+del fake_db_snapshot['metadata']
+del fake_db_snapshot['volume']
+
+
+# NOTE(andrey-mp): make Snapshot object here to check object algorithms
+fake_snapshot_obj = {
'id': '1',
'volume_id': 'fake_id',
'status': "creating",
'volume_size': 1,
'display_name': 'fake_name',
'display_description': 'fake_description',
+ 'metadata': {},
}
class TestSnapshot(test_objects.BaseObjectsTestCase):
@staticmethod
- def _compare(test, db, obj):
- for field, value in db.items():
- test.assertEqual(db[field], obj[field])
+ def _compare(test, expected, actual):
+ for field, value in expected.items():
+ test.assertEqual(expected[field], actual[field],
+ "Field '%s' is not equal" % field)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value={})
- @mock.patch('cinder.db.snapshot_get', return_value=fake_snapshot)
- def test_get_by_id(self, snapshot_get, snapshot_metadata_get):
+ @mock.patch('cinder.db.snapshot_get', return_value=fake_db_snapshot)
+ def test_get_by_id(self, snapshot_get):
snapshot = objects.Snapshot.get_by_id(self.context, 1)
- self._compare(self, fake_snapshot, snapshot)
+ self._compare(self, fake_snapshot_obj, snapshot)
def test_reset_changes(self):
snapshot = objects.Snapshot()
snapshot.obj_reset_changes(['metadata'])
self.assertEqual({'key1': 'value1'}, snapshot._orig_metadata)
- @mock.patch('cinder.db.snapshot_create', return_value=fake_snapshot)
+ @mock.patch('cinder.db.snapshot_create', return_value=fake_db_snapshot)
def test_create(self, snapshot_create):
snapshot = objects.Snapshot(context=self.context)
snapshot.create()
- self.assertEqual(fake_snapshot['id'], snapshot.id)
- self.assertEqual(fake_snapshot['volume_id'], snapshot.volume_id)
+ self.assertEqual(fake_snapshot_obj['id'], snapshot.id)
+ self.assertEqual(fake_snapshot_obj['volume_id'], snapshot.volume_id)
- @mock.patch('cinder.db.snapshot_create',
- return_value=dict(provider_id='1111-aaaa', **fake_snapshot))
+ @mock.patch('cinder.db.snapshot_create')
def test_create_with_provider_id(self, snapshot_create):
+ snapshot_create.return_value = copy.deepcopy(fake_db_snapshot)
+ snapshot_create.return_value['provider_id'] = '1111-aaaa'
+
snapshot = objects.Snapshot(context=self.context)
snapshot.create()
self.assertEqual('1111-aaaa', snapshot.provider_id)
@mock.patch('cinder.db.snapshot_update')
def test_save(self, snapshot_update):
snapshot = objects.Snapshot._from_db_object(
- self.context, objects.Snapshot(), fake_snapshot)
+ self.context, objects.Snapshot(), fake_db_snapshot)
snapshot.display_name = 'foobar'
snapshot.save()
snapshot_update.assert_called_once_with(self.context, snapshot.id,
def test_save_with_metadata(self, snapshot_update,
snapshot_metadata_update):
snapshot = objects.Snapshot._from_db_object(
- self.context, objects.Snapshot(), fake_snapshot)
+ self.context, objects.Snapshot(), fake_db_snapshot)
snapshot.display_name = 'foobar'
snapshot.metadata = {'key1': 'value1'}
self.assertEqual({'display_name': 'foobar',
@mock.patch('cinder.objects.volume.Volume.get_by_id')
def test_obj_load_attr(self, volume_get_by_id):
snapshot = objects.Snapshot._from_db_object(
- self.context, objects.Snapshot(), fake_snapshot)
+ self.context, objects.Snapshot(), fake_db_snapshot)
volume = objects.Volume(context=self.context, id=2)
volume_get_by_id.return_value = volume
self.assertEqual(volume, snapshot.volume)
@mock.patch('cinder.db.snapshot_data_get_for_project')
def test_snapshot_data_get_for_project(self, snapshot_data_get):
snapshot = objects.Snapshot._from_db_object(
- self.context, objects.Snapshot(), fake_snapshot)
+ self.context, objects.Snapshot(), fake_db_snapshot)
volume_type_id = mock.sentinel.volume_type_id
snapshot.snapshot_data_get_for_project(self.context,
self.project_id,
class TestSnapshotList(test_objects.BaseObjectsTestCase):
- @mock.patch('cinder.db.snapshot_metadata_get', return_value={})
@mock.patch('cinder.objects.volume.Volume.get_by_id')
- @mock.patch('cinder.db.snapshot_get_all', return_value=[fake_snapshot])
- def test_get_all(self, snapshot_get_all, volume_get_by_id,
- snapshot_metadata_get):
+ @mock.patch('cinder.db.snapshot_get_all', return_value=[fake_db_snapshot])
+ def test_get_all(self, snapshot_get_all, volume_get_by_id):
fake_volume_obj = fake_volume.fake_volume_obj(self.context)
volume_get_by_id.return_value = fake_volume_obj
snapshots = objects.SnapshotList.get_all(
self.context, search_opts)
self.assertEqual(1, len(snapshots))
- TestSnapshot._compare(self, fake_snapshot, snapshots[0])
+ TestSnapshot._compare(self, fake_snapshot_obj, snapshots[0])
snapshot_get_all.assert_called_once_with(self.context, search_opts)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value={})
@mock.patch('cinder.objects.Volume.get_by_id')
@mock.patch('cinder.db.snapshot_get_by_host',
- return_value=[fake_snapshot])
- def test_get_by_host(self, get_by_host, volume_get_by_id,
- snapshot_metadata_get):
+ return_value=[fake_db_snapshot])
+ def test_get_by_host(self, get_by_host, volume_get_by_id):
fake_volume_obj = fake_volume.fake_volume_obj(self.context)
volume_get_by_id.return_value = fake_volume_obj
snapshots = objects.SnapshotList.get_by_host(
self.context, 'fake-host')
self.assertEqual(1, len(snapshots))
- TestSnapshot._compare(self, fake_snapshot, snapshots[0])
+ TestSnapshot._compare(self, fake_snapshot_obj, snapshots[0])
- @mock.patch('cinder.db.snapshot_metadata_get', return_value={})
@mock.patch('cinder.objects.volume.Volume.get_by_id')
@mock.patch('cinder.db.snapshot_get_all_by_project',
- return_value=[fake_snapshot])
- def test_get_all_by_project(self, get_all_by_project, volume_get_by_id,
- snapshot_metadata_get):
+ return_value=[fake_db_snapshot])
+ def test_get_all_by_project(self, get_all_by_project, volume_get_by_id):
fake_volume_obj = fake_volume.fake_volume_obj(self.context)
volume_get_by_id.return_value = fake_volume_obj
snapshots = objects.SnapshotList.get_all_by_project(
self.context, self.project_id, search_opts)
self.assertEqual(1, len(snapshots))
- TestSnapshot._compare(self, fake_snapshot, snapshots[0])
+ TestSnapshot._compare(self, fake_snapshot_obj, snapshots[0])
get_all_by_project.assert_called_once_with(self.context,
self.project_id,
search_opts)
- @mock.patch('cinder.db.snapshot_metadata_get', return_value={})
@mock.patch('cinder.objects.volume.Volume.get_by_id')
@mock.patch('cinder.db.snapshot_get_all_for_volume',
- return_value=[fake_snapshot])
- def test_get_all_for_volume(self, get_all_for_volume, volume_get_by_id,
- snapshot_metadata_get):
+ return_value=[fake_db_snapshot])
+ def test_get_all_for_volume(self, get_all_for_volume, volume_get_by_id):
fake_volume_obj = fake_volume.fake_volume_obj(self.context)
volume_get_by_id.return_value = fake_volume_obj
snapshots = objects.SnapshotList.get_all_for_volume(
self.context, fake_volume_obj.id)
self.assertEqual(1, len(snapshots))
- TestSnapshot._compare(self, fake_snapshot, snapshots[0])
+ TestSnapshot._compare(self, fake_snapshot_obj, snapshots[0])
- @mock.patch('cinder.db.snapshot_metadata_get', return_value={})
@mock.patch('cinder.objects.volume.Volume.get_by_id')
@mock.patch('cinder.db.snapshot_get_active_by_window',
- return_value=[fake_snapshot])
+ return_value=[fake_db_snapshot])
def test_get_active_by_window(self, get_active_by_window,
- volume_get_by_id, snapshot_metadata_get):
+ volume_get_by_id):
fake_volume_obj = fake_volume.fake_volume_obj(self.context)
volume_get_by_id.return_value = fake_volume_obj
snapshots = objects.SnapshotList.get_active_by_window(
self.context, mock.sentinel.begin, mock.sentinel.end)
self.assertEqual(1, len(snapshots))
- TestSnapshot._compare(self, fake_snapshot, snapshots[0])
+ TestSnapshot._compare(self, fake_snapshot_obj, snapshots[0])
- @mock.patch('cinder.db.snapshot_metadata_get', return_value={})
@mock.patch('cinder.objects.volume.Volume.get_by_id')
@mock.patch('cinder.db.snapshot_get_all_for_cgsnapshot',
- return_value=[fake_snapshot])
+ return_value=[fake_db_snapshot])
def test_get_all_for_cgsnapshot(self, get_all_for_cgsnapshot,
- volume_get_by_id, snapshot_metadata_get):
+ volume_get_by_id):
fake_volume_obj = fake_volume.fake_volume_obj(self.context)
volume_get_by_id.return_value = fake_volume_obj
snapshots = objects.SnapshotList.get_all_for_cgsnapshot(
self.context, mock.sentinel.cgsnapshot_id)
self.assertEqual(1, len(snapshots))
- TestSnapshot._compare(self, fake_snapshot, snapshots[0])
+ TestSnapshot._compare(self, fake_snapshot_obj, snapshots[0])
+
+ @mock.patch('cinder.objects.volume.Volume.get_by_id')
+ @mock.patch('cinder.db.snapshot_get_all')
+ def test_get_all_without_metadata(self, snapshot_get_all,
+ volume_get_by_id):
+ fake_volume_obj = fake_volume.fake_volume_obj(self.context)
+ volume_get_by_id.return_value = fake_volume_obj
+
+ snapshot = copy.deepcopy(fake_db_snapshot)
+ del snapshot['snapshot_metadata']
+ snapshot_get_all.return_value = [snapshot]
+
+ search_opts = mock.sentinel.search_opts
+ self.assertRaises(exception.MetadataAbsent,
+ objects.SnapshotList.get_all,
+ self.context, search_opts)
+
+ @mock.patch('cinder.objects.volume.Volume.get_by_id')
+ @mock.patch('cinder.db.snapshot_get_all')
+ def test_get_all_with_metadata(self, snapshot_get_all, volume_get_by_id):
+ fake_volume_obj = fake_volume.fake_volume_obj(self.context)
+ volume_get_by_id.return_value = fake_volume_obj
+
+ db_snapshot = copy.deepcopy(fake_db_snapshot)
+ db_snapshot['snapshot_metadata'] = [{'key': 'fake_key',
+ 'value': 'fake_value'}]
+ snapshot_get_all.return_value = [db_snapshot]
+
+ search_opts = mock.sentinel.search_opts
+ snapshots = objects.SnapshotList.get_all(
+ self.context, search_opts)
+ self.assertEqual(1, len(snapshots))
+
+ snapshot_obj = copy.deepcopy(fake_snapshot_obj)
+ snapshot_obj['metadata'] = {'fake_key': 'fake_value'}
+ TestSnapshot._compare(self, snapshot_obj, snapshots[0])
+ snapshot_get_all.assert_called_once_with(self.context, search_opts)
Code Review / openstack-build / cinder-build.git / commitdiff