from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit.api.v2 import stubs
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder import volume
from cinder.volume import api as volume_api
res_dict = self.controller._volume_upload_image(req, id, body)
expected_res = {
'os-volume_upload_image': {
- 'id': '1',
+ 'id': fake.volume_id,
'updated_at': None,
'status': 'uploading',
'display_description': None,
--- /dev/null
+# Copyright 2016 Red Hat, Inc.
+#
+# 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.
+
+attachment_id = '4dc3bb12-ad75-41b9-ab2c-7609e743e600'
+backup_id = '707844eb-6d8a-4ac1-8b98-618e1c0b3a3a'
+volume_id = '1e5177e7-95e5-4a0f-b170-e45f4b469f6a'
+volume2_id = '43a09914-e495-475f-b862-0bda3c8918e4'
+volume3_id = '1b1cf149-219c-44ac-aee3-13121a7f86a7'
+volume_name_id = 'ee73d33c-52ed-4cb7-a8a9-2687c1205c22'
+snapshot_id = '253b2878-ec60-4793-ad19-e65496ec7aab'
+snapshot2_id = 'c02c44fa-5665-4a26-9e66-2ebaf25e5d2d'
+snapshot3_id = '454f9970-1e05-4193-a3ed-5c390c3faa18'
+cgsnapshot_id = '5e34cce3-bc97-46b7-a127-5cfb95ef445d'
+cgsnapshot2_id = '5c36d762-d6ba-4f04-bd07-88a298cc410a'
+cgsnapshot3_id = '5f392156-fc03-492a-9cb8-e46a7eedaf33'
+consistency_group_id = 'f18abf73-79ee-4f2b-8d4f-1c044148f117'
+object_id = 'd7c5b12f-d57d-4762-99ab-db5f62ae3569'
+object2_id = '51f5b8fa-c13c-48ba-8c9d-b470466cbc9c'
+object3_id = '7bf5ffa9-18a2-4b64-aab4-0798b53ee4e7'
+project_id = '89afd400-b646-4bbc-b12b-c0a4d63e5bd3'
+provider_id = '60087173-e899-470a-9e3a-ba4cffa3e3e3'
+user_id = 'c853ca26-e8ea-4797-8a52-ee124a013d0e'
+volume_type_id = '4e9e6d23-eed0-426d-b90a-28f87a94b6fe'
from oslo_versionedobjects import fields
from cinder.objects import snapshot
+from cinder.tests.unit import fake_constants as fake
def fake_db_snapshot(**updates):
db_snapshot = {
- 'id': '1',
- 'volume_id': 'fake_id',
+ 'id': fake.snapshot_id,
+ 'volume_id': fake.volume_id,
'status': "creating",
'progress': '0%',
'volume_size': 1,
from oslo_versionedobjects import fields
from cinder import objects
+from cinder.tests.unit import fake_constants as fake
def fake_db_volume(**updates):
db_volume = {
- 'id': '1',
+ 'id': fake.volume_id,
'size': 1,
- 'name': 'volume-1',
+ 'name': 'volume-%s' % fake.volume_id,
'availability_zone': 'fake_availability_zone',
'status': 'available',
'attach_status': 'detached',
def fake_db_volume_type(**updates):
db_volume_type = {
- 'id': '1',
+ 'id': fake.volume_type_id,
'name': 'type-1',
'description': 'A fake volume type',
'is_public': True,
def fake_db_volume_attachment(**updates):
db_volume_attachment = {
- 'id': '1',
- 'volume_id': '1',
+ 'id': fake.attachment_id,
+ 'volume_id': fake.volume_id,
}
for name, field in objects.VolumeAttachment.fields.items():
from cinder import exception
from cinder import objects
from cinder.objects import fields
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder.tests.unit import objects as test_objects
from cinder.tests.unit import utils
fake_backup = {
- 'id': '1',
- 'volume_id': 'fake_id',
+ 'id': fake.backup_id,
+ 'volume_id': fake.volume_id,
'status': fields.BackupStatus.CREATING,
'size': 1,
'display_name': 'fake_name',
'display_description': 'fake_description',
- 'user_id': 'fake_user',
- 'project_id': 'fake_project',
+ 'user_id': fake.user_id,
+ 'project_id': fake.project_id,
'temp_volume_id': None,
'temp_snapshot_id': None,
'snapshot_id': None,
@mock.patch('cinder.db.get_by_id', return_value=fake_backup)
def test_get_by_id(self, backup_get):
- backup = objects.Backup.get_by_id(self.context, 1)
+ backup = objects.Backup.get_by_id(self.context, fake.user_id)
self._compare(self, fake_backup, backup)
- backup_get.assert_called_once_with(self.context, models.Backup, 1)
+ backup_get.assert_called_once_with(self.context, models.Backup,
+ fake.user_id)
@mock.patch('cinder.db.sqlalchemy.api.model_query')
def test_get_by_id_no_existing_id(self, model_query):
@mock.patch('cinder.db.backup_destroy')
def test_destroy(self, backup_destroy):
- backup = objects.Backup(context=self.context, id=1)
+ backup = objects.Backup(context=self.context, id=fake.backup_id)
backup.destroy()
self.assertTrue(backup_destroy.called)
admin_context = backup_destroy.call_args[0][0]
def test_import_record(self):
utils.replace_obj_loader(self, objects.Backup)
- backup = objects.Backup(context=self.context, id=1, parent_id=None,
+ backup = objects.Backup(context=self.context, id=fake.backup_id,
+ parent_id=None,
num_dependent_backups=0)
export_string = backup.encode_record()
imported_backup = objects.Backup.decode_record(export_string)
def test_import_record_additional_info(self):
utils.replace_obj_loader(self, objects.Backup)
- backup = objects.Backup(context=self.context, id=1, parent_id=None,
+ backup = objects.Backup(context=self.context, id=fake.backup_id,
+ parent_id=None,
num_dependent_backups=0)
extra_info = {'driver': {'key1': 'value1', 'key2': 'value2'}}
extra_info_copy = extra_info.copy()
def test_import_record_additional_info_cant_overwrite(self):
utils.replace_obj_loader(self, objects.Backup)
- backup = objects.Backup(context=self.context, id=1, parent_id=None,
+ backup = objects.Backup(context=self.context, id=fake.backup_id,
+ parent_id=None,
num_dependent_backups=0)
export_string = backup.encode_record(id='fake_id')
imported_backup = objects.Backup.decode_record(export_string)
# On the second backup_get, return the backup with an updated
# display_name
backup_get.side_effect = [db_backup1, db_backup2]
- backup = objects.Backup.get_by_id(self.context, '1')
+ backup = objects.Backup.get_by_id(self.context, fake.backup_id)
self._compare(self, db_backup1, backup)
# display_name was updated, so a backup refresh should have a new value
call_bool = mock.call.__bool__()
else:
call_bool = mock.call.__nonzero__()
- backup_get.assert_has_calls([mock.call(self.context, '1'),
+ backup_get.assert_has_calls([mock.call(self.context, fake.backup_id),
call_bool,
- mock.call(self.context, '1')])
+ mock.call(self.context, fake.backup_id)])
class TestBackupList(test_objects.BaseObjectsTestCase):
from cinder.db.sqlalchemy import models
from cinder import objects
from cinder import test
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import objects as test_objects
fields = {'id': fields.UUIDField(),
'name': fields.StringField()}
- test_obj = MyTestObject(id='1', name='foo')
- refresh_obj = MyTestObject(id='1', name='bar')
+ test_obj = MyTestObject(id=fake.object_id, name='foo')
+ refresh_obj = MyTestObject(id=fake.object_id, name='bar')
with mock.patch(
'cinder.objects.base.CinderObject.get_by_id') as get_by_id:
get_by_id.return_value = refresh_obj
objects.base.CinderComparableObject):
fields = {'uuid': fields.UUIDField()}
- test_obj = MyTestObjectNoId(uuid='1', name='foo')
+ test_obj = MyTestObjectNoId(uuid=fake.object_id, name='foo')
self.assertRaises(NotImplementedError, test_obj.refresh)
from cinder import exception
from cinder import objects
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import objects as test_objects
from cinder.tests.unit.objects.test_consistencygroup import \
fake_consistencygroup
fake_cgsnapshot = {
- 'id': '1',
- 'user_id': 'fake_user_id',
- 'project_id': 'fake_project_id',
+ 'id': fake.cgsnapshot_id,
+ 'user_id': fake.user_id,
+ 'project_id': fake.project_id,
'name': 'fake_name',
'description': 'fake_description',
'status': 'creating',
- 'consistencygroup_id': 'fake_id',
+ 'consistencygroup_id': fake.consistency_group_id,
}
@mock.patch('cinder.db.sqlalchemy.api.cgsnapshot_get',
return_value=fake_cgsnapshot)
def test_get_by_id(self, cgsnapshot_get):
- cgsnapshot = objects.CGSnapshot.get_by_id(self.context, 1)
+ cgsnapshot = objects.CGSnapshot.get_by_id(self.context,
+ fake.cgsnapshot_id)
self._compare(self, fake_cgsnapshot, cgsnapshot)
@mock.patch('cinder.db.cgsnapshot_create',
self._compare(self, fake_cgsnapshot, cgsnapshot)
def test_create_with_id_except_exception(self):
- cgsnapshot = objects.CGSnapshot(context=self.context, **{'id': 2})
+ cgsnapshot = objects.CGSnapshot(context=self.context,
+ **{'id': fake.consistency_group_id})
self.assertRaises(exception.ObjectActionError, cgsnapshot.create)
@mock.patch('cinder.db.cgsnapshot_update')
@mock.patch('cinder.db.cgsnapshot_destroy')
def test_destroy(self, cgsnapshot_destroy):
- cgsnapshot = objects.CGSnapshot(context=self.context, id=1)
+ cgsnapshot = objects.CGSnapshot(context=self.context,
+ id=fake.cgsnapshot_id)
cgsnapshot.destroy()
self.assertTrue(cgsnapshot_destroy.called)
admin_context = cgsnapshot_destroy.call_args[0][0]
cgsnapshot = objects.CGSnapshot._from_db_object(
self.context, objects.CGSnapshot(), fake_cgsnapshot)
# Test consistencygroup lazy-loaded field
- consistencygroup = objects.ConsistencyGroup(context=self.context, id=2)
+ consistencygroup = objects.ConsistencyGroup(
+ context=self.context, id=fake.consistency_group_id)
consistencygroup_get_by_id.return_value = consistencygroup
self.assertEqual(consistencygroup, cgsnapshot.consistencygroup)
consistencygroup_get_by_id.assert_called_once_with(
self.context, cgsnapshot.consistencygroup_id)
# Test snapshots lazy-loaded field
snapshots_objs = [objects.Snapshot(context=self.context, id=i)
- for i in [3, 4, 5]]
+ for i in [fake.snapshot_id, fake.snapshot2_id,
+ fake.snapshot3_id]]
snapshots = objects.SnapshotList(context=self.context,
objects=snapshots_objs)
snapshotlist_get_for_cgs.return_value = snapshots
# On the second cgsnapshot_get, return the CGSnapshot with an updated
# description
cgsnapshot_get.side_effect = [db_cgsnapshot1, db_cgsnapshot2]
- cgsnapshot = objects.CGSnapshot.get_by_id(self.context, '1')
+ cgsnapshot = objects.CGSnapshot.get_by_id(self.context,
+ fake.cgsnapshot_id)
self._compare(self, db_cgsnapshot1, cgsnapshot)
# description was updated, so a CGSnapshot refresh should have a new
call_bool = mock.call.__bool__()
else:
call_bool = mock.call.__nonzero__()
- cgsnapshot_get.assert_has_calls([mock.call(self.context, '1'),
+ cgsnapshot_get.assert_has_calls([mock.call(self.context,
+ fake.cgsnapshot_id),
call_bool,
- mock.call(self.context, '1')])
+ mock.call(self.context,
+ fake.cgsnapshot_id)])
class TestCGSnapshotList(test_objects.BaseObjectsTestCase):
from cinder import exception
from cinder import objects
from cinder.objects import fields
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import objects as test_objects
fake_consistencygroup = {
- 'id': '1',
- 'user_id': 'fake_user_id',
- 'project_id': 'fake_project_id',
+ 'id': fake.consistency_group_id,
+ 'user_id': fake.user_id,
+ 'project_id': fake.project_id,
'host': 'fake_host',
'availability_zone': 'fake_az',
'name': 'fake_name',
'description': 'fake_description',
- 'volume_type_id': 'fake_volume_type_id',
+ 'volume_type_id': fake.volume_type_id,
'status': fields.ConsistencyGroupStatus.CREATING,
- 'cgsnapshot_id': 'fake_id',
+ 'cgsnapshot_id': fake.cgsnapshot_id,
'source_cgid': None,
}
fake_cgsnapshot = {
- 'id': '1',
- 'user_id': 'fake_user_id',
- 'project_id': 'fake_project_id',
+ 'id': fake.cgsnapshot_id,
+ 'user_id': fake.user_id,
+ 'project_id': fake.project_id,
'name': 'fake_name',
'description': 'fake_description',
'status': 'creating',
- 'consistencygroup_id': 'fake_id',
+ 'consistencygroup_id': fake.consistency_group_id,
}
@mock.patch('cinder.db.sqlalchemy.api.consistencygroup_get',
return_value=fake_consistencygroup)
def test_get_by_id(self, consistencygroup_get):
- consistencygroup = objects.ConsistencyGroup.get_by_id(self.context, 1)
+ consistencygroup = objects.ConsistencyGroup.get_by_id(
+ self.context, fake.consistency_group_id)
self._compare(self, fake_consistencygroup, consistencygroup)
- consistencygroup_get.assert_called_once_with(self.context, 1)
+ consistencygroup_get.assert_called_once_with(
+ self.context, fake.consistency_group_id)
@mock.patch('cinder.db.sqlalchemy.api.model_query')
def test_get_by_id_no_existing_id(self, model_query):
self._compare(self, fake_consistencygroup, consistencygroup)
def test_create_with_id_except_exception(self, ):
- consistencygroup = objects.ConsistencyGroup(context=self.context,
- **{'id': 1})
+ consistencygroup = objects.ConsistencyGroup(
+ context=self.context, **{'id': fake.consistency_group_id})
self.assertRaises(exception.ObjectActionError, consistencygroup.create)
@mock.patch('cinder.db.consistencygroup_update')
consistencygroup = objects.ConsistencyGroup._from_db_object(
self.context, objects.ConsistencyGroup(), fake_consistencygroup)
cgsnapshots_objs = [objects.CGSnapshot(context=self.context, id=i)
- for i in [3, 4, 5]]
+ for i in [fake.cgsnapshot_id, fake.cgsnapshot2_id,
+ fake.cgsnapshot3_id]]
cgsnapshots = objects.CGSnapshotList(objects=cgsnapshots_objs)
consistencygroup.name = 'foobar'
consistencygroup.cgsnapshots = cgsnapshots
consistencygroup = objects.ConsistencyGroup._from_db_object(
self.context, objects.ConsistencyGroup(), fake_consistencygroup)
volumes_objs = [objects.Volume(context=self.context, id=i)
- for i in [3, 4, 5]]
+ for i in [fake.volume_id, fake.volume2_id,
+ fake.volume3_id]]
volumes = objects.VolumeList(objects=volumes_objs)
consistencygroup.name = 'foobar'
consistencygroup.volumes = volumes
self.context, objects.ConsistencyGroup(), fake_consistencygroup)
# Test cgsnapshots lazy-loaded field
cgsnapshots_objs = [objects.CGSnapshot(context=self.context, id=i)
- for i in [3, 4, 5]]
+ for i in [fake.cgsnapshot_id, fake.cgsnapshot2_id,
+ fake.cgsnapshot3_id]]
cgsnapshots = objects.CGSnapshotList(context=self.context,
objects=cgsnapshots_objs)
mock_cgsnap_get_all_by_group.return_value = cgsnapshots
# Test volumes lazy-loaded field
volume_objs = [objects.Volume(context=self.context, id=i)
- for i in [3, 4, 5]]
+ for i in [fake.volume_id, fake.volume2_id,
+ fake.volume3_id]]
volumes = objects.VolumeList(context=self.context, objects=volume_objs)
mock_vol_get_all_by_group.return_value = volumes
self.assertEqual(volumes, consistencygroup.volumes)
@mock.patch('cinder.db.consistencygroup_destroy')
def test_destroy(self, consistencygroup_destroy):
- consistencygroup = objects.ConsistencyGroup(context=self.context,
- id='1')
+ consistencygroup = objects.ConsistencyGroup(
+ context=self.context, id=fake.consistency_group_id)
consistencygroup.destroy()
self.assertTrue(consistencygroup_destroy.called)
admin_context = consistencygroup_destroy.call_args[0][0]
# On the second consistencygroup_get, return the ConsistencyGroup with
# an updated description
consistencygroup_get.side_effect = [db_cg1, db_cg2]
- cg = objects.ConsistencyGroup.get_by_id(self.context, '1')
+ cg = objects.ConsistencyGroup.get_by_id(self.context,
+ fake.consistency_group_id)
self._compare(self, db_cg1, cg)
# description was updated, so a ConsistencyGroup refresh should have a
call_bool = mock.call.__bool__()
else:
call_bool = mock.call.__nonzero__()
- consistencygroup_get.assert_has_calls([mock.call(self.context, '1'),
- call_bool,
- mock.call(self.context, '1')])
+ consistencygroup_get.assert_has_calls([
+ mock.call(
+ self.context,
+ fake.consistency_group_id),
+ call_bool,
+ mock.call(
+ self.context,
+ fake.consistency_group_id)])
class TestConsistencyGroupList(test_objects.BaseObjectsTestCase):
from cinder.db.sqlalchemy import models
from cinder import exception
from cinder import objects
+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
fake_db_snapshot = fake_snapshot.fake_db_snapshot(
- cgsnapshot_id='fake_cgsnap_id')
+ cgsnapshot_id=fake.cgsnapshot_id)
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',
+ 'id': fake.snapshot_id,
+ 'volume_id': fake.volume_id,
'status': "creating",
'progress': '0%',
'volume_size': 1,
@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_create.return_value['provider_id'] = fake.provider_id
snapshot = objects.Snapshot(context=self.context)
snapshot.create()
- self.assertEqual('1111-aaaa', snapshot.provider_id)
+ self.assertEqual(fake.provider_id, snapshot.provider_id)
@mock.patch('cinder.db.snapshot_update')
def test_save(self, snapshot_update):
snapshot.save()
snapshot_update.assert_called_once_with(self.context, snapshot.id,
{'display_name': 'foobar'})
- snapshot_metadata_update.assert_called_once_with(self.context, '1',
+ snapshot_metadata_update.assert_called_once_with(self.context,
+ fake.snapshot_id,
{'key1': 'value1'},
True)
@mock.patch('cinder.db.snapshot_destroy')
def test_destroy(self, snapshot_destroy):
- snapshot = objects.Snapshot(context=self.context, id=1)
+ snapshot = objects.Snapshot(context=self.context, id=fake.snapshot_id)
snapshot.destroy()
- snapshot_destroy.assert_called_once_with(self.context, '1')
+ snapshot_destroy.assert_called_once_with(self.context,
+ fake.snapshot_id)
@mock.patch('cinder.db.snapshot_metadata_delete')
def test_delete_metadata_key(self, snapshot_metadata_delete):
- snapshot = objects.Snapshot(self.context, id=1)
+ snapshot = objects.Snapshot(self.context, id=fake.snapshot_id)
snapshot.metadata = {'key1': 'value1', 'key2': 'value2'}
self.assertEqual({}, snapshot._orig_metadata)
snapshot.delete_metadata_key(self.context, 'key2')
self.assertEqual({'key1': 'value1'}, snapshot.metadata)
- snapshot_metadata_delete.assert_called_once_with(self.context, '1',
+ snapshot_metadata_delete.assert_called_once_with(self.context,
+ fake.snapshot_id,
'key2')
def test_obj_fields(self):
- volume = objects.Volume(context=self.context, id=2, _name_id=2)
- snapshot = objects.Snapshot(context=self.context, id=1,
+ volume = objects.Volume(context=self.context, id=fake.volume_id,
+ _name_id=fake.volume_name_id)
+ snapshot = objects.Snapshot(context=self.context, id=fake.volume_id,
volume=volume)
self.assertEqual(['name', 'volume_name'], snapshot.obj_extra_fields)
- self.assertEqual('snapshot-1', snapshot.name)
- self.assertEqual('volume-2', snapshot.volume_name)
+ self.assertEqual('snapshot-%s' % fake.volume_id, snapshot.name)
+ self.assertEqual('volume-%s' % fake.volume_name_id,
+ snapshot.volume_name)
@mock.patch('cinder.objects.volume.Volume.get_by_id')
@mock.patch('cinder.objects.cgsnapshot.CGSnapshot.get_by_id')
snapshot = objects.Snapshot._from_db_object(
self.context, objects.Snapshot(), fake_db_snapshot)
# Test volume lazy-loaded field
- volume = objects.Volume(context=self.context, id=2)
+ volume = objects.Volume(context=self.context, id=fake.volume_id)
volume_get_by_id.return_value = volume
self.assertEqual(volume, snapshot.volume)
volume_get_by_id.assert_called_once_with(self.context,
snapshot.volume_id)
# Test cgsnapshot lazy-loaded field
- cgsnapshot = objects.CGSnapshot(context=self.context, id=2)
+ cgsnapshot = objects.CGSnapshot(context=self.context,
+ id=fake.cgsnapshot_id)
cgsnapshot_get_by_id.return_value = cgsnapshot
self.assertEqual(cgsnapshot, snapshot.cgsnapshot)
cgsnapshot_get_by_id.assert_called_once_with(self.context,
# On the second snapshot_get, return the snapshot with an updated
# display_name
snapshot_get.side_effect = [db_snapshot1, db_snapshot2]
- snapshot = objects.Snapshot.get_by_id(self.context, '1')
+ snapshot = objects.Snapshot.get_by_id(self.context, fake.snapshot_id)
self._compare(self, db_snapshot1, snapshot)
# display_name was updated, so a snapshot refresh should have a new
call_bool = mock.call.__bool__()
else:
call_bool = mock.call.__nonzero__()
- snapshot_get.assert_has_calls([mock.call(self.context, '1'),
- call_bool,
- mock.call(self.context, '1')])
+ snapshot_get.assert_has_calls([
+ mock.call(self.context,
+ fake.snapshot_id),
+ call_bool,
+ mock.call(self.context,
+ fake.snapshot_id)])
class TestSnapshotList(test_objects.BaseObjectsTestCase):
from cinder import exception
from cinder import objects
from cinder.tests.unit 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
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, 1)
- volume_get.assert_called_once_with(self.context, 1)
+ 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')
self.assertTrue(admin_context.is_admin)
def test_obj_fields(self):
- volume = objects.Volume(context=self.context, id=2, _name_id=2)
+ 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-2', volume.name)
- self.assertEqual('2', volume.name_id)
+ 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,
@mock.patch('cinder.db.volume_metadata_delete')
def test_delete_metadata_key(self, metadata_delete):
- volume = objects.Volume(self.context, id=1)
+ 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, '1', 'key2')
+ 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')
# Case2. volume2.volume_type_id = 1
fake2 = fake_volume.fake_db_volume()
- fake2.update({'volume_type_id': 1})
+ 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=1)
+ 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=2)
+ 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=2)
+ 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,
# Test volume_attachment lazy-loaded field
va_objs = [objects.VolumeAttachment(context=self.context, id=i)
- for i in [3, 4, 5]]
+ 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
# 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, '1')
+ 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
call_bool = mock.call.__bool__()
else:
call_bool = mock.call.__nonzero__()
- volume_get.assert_has_calls([mock.call(self.context, '1'),
+ volume_get.assert_has_calls([mock.call(self.context, fake.volume_id),
call_bool,
- mock.call(self.context, '1')])
+ mock.call(self.context, fake.volume_id)])
def test_metadata_aliases(self):
volume = objects.Volume(context=self.context)
@mock.patch('cinder.db.volume_metadata_update', return_value={})
@mock.patch('cinder.db.volume_update')
def test_finish_volume_migration(self, volume_update, metadata_update):
- src_volume_db = fake_volume.fake_db_volume(**{'id': '1'})
- dest_volume_db = fake_volume.fake_db_volume(**{'id': '2'})
+ src_volume_db = fake_volume.fake_db_volume(**{'id': fake.volume_id})
+ dest_volume_db = fake_volume.fake_db_volume(**{'id': fake.volume2_id})
src_volume = objects.Volume._from_db_object(
self.context, objects.Volume(), src_volume_db,
expected_attrs=['metadata', 'glance_metadata'])
import six
from cinder import objects
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder.tests.unit import objects as test_objects
def test_get_by_id(self, volume_attachment_get):
db_attachment = fake_volume.fake_db_volume_attachment()
volume_attachment_get.return_value = db_attachment
- attachment = objects.VolumeAttachment.get_by_id(self.context, '1')
+ attachment = objects.VolumeAttachment.get_by_id(self.context,
+ fake.attachment_id)
self._compare(self, db_attachment, attachment)
@mock.patch('cinder.db.volume_attachment_update')
# On the second volume_attachment_get, return the volume attachment
# with an updated mountpoint
attachment_get.side_effect = [db_attachment1, db_attachment2]
- attachment = objects.VolumeAttachment.get_by_id(self.context, '1')
+ attachment = objects.VolumeAttachment.get_by_id(self.context,
+ fake.attachment_id)
self._compare(self, db_attachment1, attachment)
# mountpoint was updated, so a volume attachment refresh should have a
call_bool = mock.call.__bool__()
else:
call_bool = mock.call.__nonzero__()
- attachment_get.assert_has_calls([mock.call(self.context, '1'),
+ attachment_get.assert_has_calls([mock.call(self.context,
+ fake.attachment_id),
call_bool,
- mock.call(self.context, '1')])
+ mock.call(self.context,
+ fake.attachment_id)])
class TestVolumeAttachmentList(test_objects.BaseObjectsTestCase):
import six
from cinder import objects
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder.tests.unit import objects as test_objects
def test_get_by_id(self, volume_type_get):
db_volume_type = fake_volume.fake_db_volume_type()
volume_type_get.return_value = db_volume_type
- volume_type = objects.VolumeType.get_by_id(self.context, '1')
+ volume_type = objects.VolumeType.get_by_id(self.context,
+ fake.volume_type_id)
self._compare(self, db_volume_type, volume_type)
@mock.patch('cinder.volume.volume_types.create')
db_type2 = db_type1.copy()
db_type2['description'] = 'foobar'
- # On the second _volume_type_get_full, return the volume type with an
# updated description
volume_type_get.side_effect = [db_type1, db_type2]
- volume_type = objects.VolumeType.get_by_id(self.context, '1')
+ volume_type = objects.VolumeType.get_by_id(self.context,
+ fake.volume_type_id)
self._compare(self, db_type1, volume_type)
# description was updated, so a volume type refresh should have a new
call_bool = mock.call.__bool__()
else:
call_bool = mock.call.__nonzero__()
- volume_type_get.assert_has_calls([mock.call(self.context, '1'),
+ volume_type_get.assert_has_calls([mock.call(self.context,
+ fake.volume_type_id),
call_bool,
- mock.call(self.context, '1')])
+ mock.call(self.context,
+ fake.volume_type_id)])
class TestVolumeTypeList(test_objects.BaseObjectsTestCase):
from cinder import context
from cinder import exception
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder.tests.unit.volume.drivers.emc import scaleio
from cinder.tests.unit.volume.drivers.emc.scaleio import mocks
test_vol = self.driver.update_migrated_volume(
self.ctx, self.volume, self.new_volume, 'unavailable')
self.assertFalse(mock_rename.called)
- self.assertEqual({'_name_id': '1', 'provider_location': None},
+ self.assertEqual({'_name_id': fake.volume_id,
+ 'provider_location': None},
test_vol)
@mock.patch(