object2_id = '51f5b8fa-c13c-48ba-8c9d-b470466cbc9c'
object3_id = '7bf5ffa9-18a2-4b64-aab4-0798b53ee4e7'
project_id = '89afd400-b646-4bbc-b12b-c0a4d63e5bd3'
+project2_id = '452ebfbc-55d9-402a-87af-65061916c24b'
provider_id = '60087173-e899-470a-9e3a-ba4cffa3e3e3'
snapshot_id = '253b2878-ec60-4793-ad19-e65496ec7aab'
snapshot2_id = 'c02c44fa-5665-4a26-9e66-2ebaf25e5d2d'
snapshot3_id = '454f9970-1e05-4193-a3ed-5c390c3faa18'
user_id = 'c853ca26-e8ea-4797-8a52-ee124a013d0e'
+user2_id = '95f7b7ed-bd7f-426e-b05f-f1ffeb4f09df'
volume_id = '1e5177e7-95e5-4a0f-b170-e45f4b469f6a'
volume2_id = '43a09914-e495-475f-b862-0bda3c8918e4'
volume3_id = '1b1cf149-219c-44ac-aee3-13121a7f86a7'
from cinder import exception
from cinder import test
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
"data_transferred": 318
}]
REPLICATED_VOLUME_OBJS = [
- fake_volume.fake_volume_obj(None, id='repl-1'),
- fake_volume.fake_volume_obj(None, id='repl-2'),
- fake_volume.fake_volume_obj(None, id='repl-3'),
+ fake_volume.fake_volume_obj(None, id=fake.volume_id),
+ fake_volume.fake_volume_obj(None, id=fake.volume2_id),
+ fake_volume.fake_volume_obj(None, id=fake.volume3_id),
]
REPLICATED_VOLUME_SNAPS = [
{
- "source": "array1:volume-repl-1-cinder",
+ "source": "array1:volume-%s-cinder" % fake.volume_id,
"serial": "BBA481C01639104E0001D5F7",
"created": "2014-12-04T22:59:38Z",
- "name": "array1:cinder-repl-pg.2.volume-repl-1-cinder",
+ "name": "array1:cinder-repl-pg.2.volume-%s-cinder" % fake.volume_id,
"size": 1048576
},
{
- "source": "array1:volume-repl-2-cinder",
+ "source": "array1:volume-%s-cinder" % fake.volume2_id,
"serial": "BBA481C01639104E0001D5F8",
"created": "2014-12-04T22:59:38Z",
- "name": "array1:cinder-repl-pg.2.volume-repl-2-cinder",
+ "name": "array1:cinder-repl-pg.2.volume-%s-cinder" % fake.volume2_id,
"size": 1048576
},
{
- "source": "array1:volume-repl-3-cinder",
+ "source": "array1:volume-%s-cinder" % fake.volume3_id,
"serial": "BBA481C01639104E0001D5F9",
"created": "2014-12-04T22:59:38Z",
- "name": "array1:cinder-repl-pg.2.volume-repl-3-cinder",
+ "name": "array1:cinder-repl-pg.2.volume-%s-cinder" % fake.volume3_id,
"size": 1048576
}
]
# unnecessary attributes should be removed from image volume
vol.consistencygroup = None
result = self.volume._clone_image_volume(self.context, vol,
- {'id': 'fake'})
+ {'id': fake.volume_id})
self.assertNotEqual(False, result)
mock_reserve.assert_called_once_with(self.context, volumes=1,
mock_commit, mock_rollback):
vol = tests_utils.create_volume(self.context, **self.volume_params)
with mock.patch.object(objects, 'Volume', side_effect=ValueError):
- self.assertFalse(self.volume._clone_image_volume(self.context, vol,
- {'id': 'fake'}))
+ self.assertFalse(self.volume._clone_image_volume(
+ self.context, vol, {'id': fake.volume_id}))
mock_reserve.assert_called_once_with(self.context, volumes=1,
gigabytes=vol.size)
with mock.patch.object(self.volume.driver,
'migrate_volume') as mock_migrate_volume:
mock_migrate_volume.side_effect = (
- lambda x, y, z, new_type_id=None: (True, {'user_id': 'foo'}))
+ lambda x, y, z, new_type_id=None: (
+ True, {'user_id': fake.user_id}))
self.volume.migrate_volume(self.context, volume.id, host_obj,
False, volume=volume)
self.assertEqual('newhost', volume.host)
fake_new_host = 'fake_new_host'
fake_update = {'_name_id': fake.volume2_name_id,
'provider_location': 'updated_location'}
- fake_elevated = context.RequestContext('fake', self.project_id,
+ fake_elevated = context.RequestContext(fake.user_id, self.project_id,
is_admin=True)
volume = tests_utils.create_volume(self.context, size=1,
status='available',
self.assertEqual(temp_vol.id, backup_obj.temp_volume_id)
def test__create_temp_volume_from_snapshot(self):
- volume_dict = {'id': 'fake',
+ volume_dict = {'id': fake.snapshot_id,
'host': 'fakehost',
'availability_zone': 'fakezone',
'size': 1}
def test_retype_volume(self):
vol = tests_utils.create_volume(self.context)
- new_type = 'fake'
+ new_type = fake.volume_type_id
diff = {}
host = 'fake_host'
retyped = self.volume.driver.retype(self.context, vol, new_type,
from cinder import exception
from cinder import objects
from cinder import test
+from cinder.tests.unit import fake_constants as fake
class VolumeGlanceMetadataTestCase(test.TestCase):
ctxt = context.get_admin_context()
self.assertRaises(exception.VolumeNotFound,
db.volume_glance_metadata_create,
- ctxt, 1, 'key1', 'value1')
+ ctxt, fake.volume_id, 'key1', 'value1')
self.assertRaises(exception.VolumeNotFound,
- db.volume_glance_metadata_get, ctxt, 1)
- db.volume_glance_metadata_delete_by_volume(ctxt, 10)
+ db.volume_glance_metadata_get, ctxt, fake.volume_id)
+ db.volume_glance_metadata_delete_by_volume(ctxt, fake.volume2_id)
def test_vol_update_glance_metadata(self):
ctxt = context.get_admin_context()
- db.volume_create(ctxt, {'id': 1})
- db.volume_create(ctxt, {'id': 2})
- db.volume_glance_metadata_create(ctxt, 1, 'key1', 'value1')
- db.volume_glance_metadata_create(ctxt, 2, 'key1', 'value1')
- db.volume_glance_metadata_create(ctxt, 2, 'key2', 'value2')
- db.volume_glance_metadata_create(ctxt, 2, 'key3', 123)
-
- expected_metadata_1 = {'volume_id': '1',
+ db.volume_create(ctxt, {'id': fake.volume_id})
+ db.volume_create(ctxt, {'id': fake.volume2_id})
+ db.volume_glance_metadata_create(ctxt, fake.volume_id, 'key1',
+ 'value1')
+ db.volume_glance_metadata_create(ctxt, fake.volume2_id, 'key1',
+ 'value1')
+ db.volume_glance_metadata_create(ctxt, fake.volume2_id, 'key2',
+ 'value2')
+ db.volume_glance_metadata_create(ctxt, fake.volume2_id, 'key3', 123)
+
+ expected_metadata_1 = {'volume_id': fake.volume_id,
'key': 'key1',
'value': 'value1'}
- metadata = db.volume_glance_metadata_get(ctxt, 1)
+ metadata = db.volume_glance_metadata_get(ctxt, fake.volume_id)
self.assertEqual(1, len(metadata))
for key, value in expected_metadata_1.items():
self.assertEqual(value, metadata[0][key])
- expected_metadata_2 = ({'volume_id': '2',
+ expected_metadata_2 = ({'volume_id': fake.volume2_id,
'key': 'key1',
'value': 'value1'},
- {'volume_id': '2',
+ {'volume_id': fake.volume2_id,
'key': 'key2',
'value': 'value2'},
- {'volume_id': '2',
+ {'volume_id': fake.volume2_id,
'key': 'key3',
'value': '123'})
- metadata = db.volume_glance_metadata_get(ctxt, 2)
+ metadata = db.volume_glance_metadata_get(ctxt, fake.volume2_id)
self.assertEqual(3, len(metadata))
for expected, meta in zip(expected_metadata_2, metadata):
for key, value in expected.items():
self.assertRaises(exception.GlanceMetadataExists,
db.volume_glance_metadata_create,
- ctxt, 1, 'key1', 'value1a')
+ ctxt, fake.volume_id, 'key1', 'value1a')
- metadata = db.volume_glance_metadata_get(ctxt, 1)
+ metadata = db.volume_glance_metadata_get(ctxt, fake.volume_id)
self.assertEqual(1, len(metadata))
for key, value in expected_metadata_1.items():
self.assertEqual(value, metadata[0][key])
def test_vols_get_glance_metadata(self):
ctxt = context.get_admin_context()
- db.volume_create(ctxt, {'id': '1'})
- db.volume_create(ctxt, {'id': '2'})
+ db.volume_create(ctxt, {'id': fake.volume_id})
+ db.volume_create(ctxt, {'id': fake.volume2_id})
db.volume_create(ctxt, {'id': '3'})
- db.volume_glance_metadata_create(ctxt, '1', 'key1', 'value1')
- db.volume_glance_metadata_create(ctxt, '2', 'key2', 'value2')
- db.volume_glance_metadata_create(ctxt, '2', 'key22', 'value22')
+ db.volume_glance_metadata_create(ctxt, fake.volume_id, 'key1',
+ 'value1')
+ db.volume_glance_metadata_create(ctxt, fake.volume2_id, 'key2',
+ 'value2')
+ db.volume_glance_metadata_create(ctxt, fake.volume2_id, 'key22',
+ 'value22')
metadata = db.volume_glance_metadata_get_all(ctxt)
self.assertEqual(3, len(metadata))
- self._assert_metadata_equals('1', 'key1', 'value1', metadata[0])
- self._assert_metadata_equals('2', 'key2', 'value2', metadata[1])
- self._assert_metadata_equals('2', 'key22', 'value22', metadata[2])
+ self._assert_metadata_equals(fake.volume_id, 'key1', 'value1',
+ metadata[0])
+ self._assert_metadata_equals(fake.volume2_id, 'key2', 'value2',
+ metadata[1])
+ self._assert_metadata_equals(fake.volume2_id, 'key22', 'value22',
+ metadata[2])
def _assert_metadata_equals(self, volume_id, key, value, observed):
self.assertEqual(volume_id, observed.volume_id)
def test_vol_delete_glance_metadata(self):
ctxt = context.get_admin_context()
- db.volume_create(ctxt, {'id': 1})
- db.volume_glance_metadata_delete_by_volume(ctxt, 1)
- db.volume_glance_metadata_create(ctxt, 1, 'key1', 'value1')
- db.volume_glance_metadata_delete_by_volume(ctxt, 1)
+ db.volume_create(ctxt, {'id': fake.volume_id})
+ db.volume_glance_metadata_delete_by_volume(ctxt, fake.volume_id)
+ db.volume_glance_metadata_create(ctxt, fake.volume_id, 'key1',
+ 'value1')
+ db.volume_glance_metadata_delete_by_volume(ctxt, fake.volume_id)
self.assertRaises(exception.GlanceMetadataNotFound,
- db.volume_glance_metadata_get, ctxt, 1)
+ db.volume_glance_metadata_get, ctxt, fake.volume_id)
def test_vol_glance_metadata_copy_to_snapshot(self):
ctxt = context.get_admin_context()
- db.volume_create(ctxt, {'id': 1})
- snap = objects.Snapshot(ctxt, volume_id=1)
+ db.volume_create(ctxt, {'id': fake.volume_id})
+ snap = objects.Snapshot(ctxt, volume_id=fake.volume_id)
snap.create()
- db.volume_glance_metadata_create(ctxt, 1, 'key1', 'value1')
- db.volume_glance_metadata_copy_to_snapshot(ctxt, snap.id, 1)
+ db.volume_glance_metadata_create(ctxt, fake.volume_id, 'key1',
+ 'value1')
+ db.volume_glance_metadata_copy_to_snapshot(ctxt, snap.id,
+ fake.volume_id)
expected_meta = {'snapshot_id': snap.id,
'key': 'key1',
def test_vol_glance_metadata_copy_from_volume_to_volume(self):
ctxt = context.get_admin_context()
- db.volume_create(ctxt, {'id': 1})
- db.volume_create(ctxt, {'id': 100, 'source_volid': 1})
- db.volume_glance_metadata_create(ctxt, 1, 'key1', 'value1')
- db.volume_glance_metadata_copy_from_volume_to_volume(ctxt, 1, 100)
+ db.volume_create(ctxt, {'id': fake.volume_id})
+ db.volume_create(ctxt, {'id': fake.volume2_id,
+ 'source_volid': fake.volume_id})
+ db.volume_glance_metadata_create(ctxt, fake.volume_id, 'key1',
+ 'value1')
+ db.volume_glance_metadata_copy_from_volume_to_volume(ctxt,
+ fake.volume_id,
+ fake.volume2_id)
expected_meta = {'key': 'key1',
'value': 'value1'}
- for meta in db.volume_glance_metadata_get(ctxt, 100):
+ for meta in db.volume_glance_metadata_get(ctxt, fake.volume2_id):
for (key, value) in expected_meta.items():
self.assertEqual(value, meta[key])
from cinder import objects
from cinder import test
from cinder.tests.unit import fake_backup
+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 utils as tests_utils
def test_manage_existing_snapshot(self):
volume_update = {'host': 'fake_host'}
snpshot = {
- 'id': 1,
- 'volume_id': 'fake_id',
+ 'id': fake.snapshot_id,
+ 'volume_id': fake.volume_id,
'status': "creating",
'progress': '0%',
'volume_size': 0,
from cinder import objects
from cinder import quota
from cinder import test
+from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import utils
from cinder.transfer import api as transfer_api
"""Test cases for volume transfer code."""
def setUp(self):
super(VolumeTransferTestCase, self).setUp()
- self.ctxt = context.RequestContext(user_id='user_id',
- project_id='project_id')
+ self.ctxt = context.RequestContext(user_id=fake.user_id,
+ project_id=fake.project_id)
self.updated_at = datetime.datetime(1, 1, 1, 1, 1, 1)
@mock.patch('cinder.volume.utils.notify_about_volume_usage')
svc = self.start_service('volume', host='test_host')
self.addCleanup(svc.stop)
tx_api = transfer_api.API()
- volume = utils.create_volume(self.ctxt, volume_type_id='12345',
+ volume = utils.create_volume(self.ctxt,
+ volume_type_id=fake.volume_type_id,
updated_at=self.updated_at)
transfer = tx_api.create(self.ctxt, volume.id, 'Description')
- self.ctxt.user_id = 'new_user_id'
- self.ctxt.project_id = 'new_project_id'
+ self.ctxt.user_id = fake.user2_id
+ self.ctxt.project_id = fake.project2_id
response = tx_api.accept(self.ctxt,
transfer['id'],
transfer['auth_key'])
volume = objects.Volume.get_by_id(self.ctxt, volume.id)
- self.assertEqual('new_project_id', volume.project_id)
- self.assertEqual('new_user_id', volume.user_id)
+ self.assertEqual(fake.project2_id, volume.project_id)
+ self.assertEqual(fake.user2_id, volume.user_id)
self.assertEqual(response['volume_id'], volume.id,
'Unexpected volume id in response.')
# QUOTAS.add_volume_type_opts
reserve_opt = {'volumes': 1, 'gigabytes': 1}
release_opt = {'volumes': -1, 'gigabytes': -1}
- calls = [mock.call(self.ctxt, reserve_opt, '12345'),
- mock.call(self.ctxt, release_opt, '12345')]
+ calls = [mock.call(self.ctxt, reserve_opt, fake.volume_type_id),
+ mock.call(self.ctxt, release_opt, fake.volume_type_id)]
mock_quota_voltype.assert_has_calls(calls)
# QUOTAS.reserve
calls = [mock.call(mock.ANY, **reserve_opt),
- mock.call(mock.ANY, project_id='project_id', **release_opt)]
+ mock.call(mock.ANY, project_id=fake.project_id,
+ **release_opt)]
mock_quota_reserve.assert_has_calls(calls)
@mock.patch.object(QUOTAS, "reserve")
svc = self.start_service('volume', host='test_host')
self.addCleanup(svc.stop)
tx_api = transfer_api.API()
- volume = utils.create_volume(self.ctxt, volume_type_id='12345',
+ volume = utils.create_volume(self.ctxt,
+ volume_type_id=fake.volume_type_id,
updated_at=self.updated_at)
transfer = tx_api.create(self.ctxt, volume.id, 'Description')
fake_overs = ['volumes_lvmdriver-3']
quotas=fake_quotas,
usages=fake_usages)
- self.ctxt.user_id = 'new_user_id'
- self.ctxt.project_id = 'new_project_id'
+ self.ctxt.user_id = fake.user2_id
+ self.ctxt.project_id = fake.project2_id
self.assertRaises(exception.VolumeLimitExceeded,
tx_api.accept,
self.ctxt,
ts = tx_api.get_all(self.ctxt)
self.assertEqual(1, len(ts), 'Unexpected number of transfers.')
- nctxt = context.RequestContext(user_id='new_user_id',
- project_id='new_project_id')
+ nctxt = context.RequestContext(user_id=fake.user2_id,
+ project_id=fake.project2_id)
utils.create_volume(nctxt, updated_at=self.updated_at)
self.assertRaises(exception.TransferNotFound,
tx_api.get,
from cinder import context
from cinder import exception
from cinder import test
+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 import utils
@mock.patch('cinder.objects.Volume.get_by_id')
def test_usage_from_snapshot(self, volume_get_by_id):
raw_volume = {
- 'id': '55614621',
+ 'id': fake.volume_id,
'availability_zone': 'nova'
}
ctxt = context.get_admin_context()
volume_obj = fake_volume.fake_volume_obj(ctxt, **raw_volume)
volume_get_by_id.return_value = volume_obj
raw_snapshot = {
- 'project_id': '12b0330ec2584a',
- 'user_id': '158cba1b8c2bb6008e',
+ 'project_id': fake.project_id,
+ 'user_id': fake.user_id,
'volume': volume_obj,
- 'volume_id': '55614621',
+ 'volume_id': fake.volume_id,
'volume_size': 1,
- 'id': '343434a2',
+ 'id': fake.snapshot_id,
'display_name': '11',
'created_at': '2014-12-11T10:10:00',
'status': 'pause',
snapshot_obj = fake_snapshot.fake_snapshot_obj(ctxt, **raw_snapshot)
usage_info = volume_utils._usage_from_snapshot(snapshot_obj)
expected_snapshot = {
- 'tenant_id': '12b0330ec2584a',
- 'user_id': '158cba1b8c2bb6008e',
+ 'tenant_id': fake.project_id,
+ 'user_id': fake.user_id,
'availability_zone': 'nova',
- 'volume_id': '55614621',
+ 'volume_id': fake.volume_id,
'volume_size': 1,
- 'snapshot_id': '343434a2',
+ 'snapshot_id': fake.snapshot_id,
'display_name': '11',
'created_at': 'DONTCARE',
'status': 'pause',