return IMPL.volume_get_all_by_host(context, host)
-def volume_get_all_by_instance_uuid(context, instance_uuid):
- """Get all volumes belonging to a instance."""
- return IMPL.volume_get_all_by_instance_uuid(context, instance_uuid)
-
-
def volume_get_all_by_project(context, project_id, marker, limit, sort_key,
sort_dir, filters=None):
"""Get all volumes belonging to a project."""
return _volume_get_query(context).filter_by(host=host).all()
-@require_admin_context
-def volume_get_all_by_instance_uuid(context, instance_uuid):
- result = model_query(context, models.Volume, read_deleted="no").\
- options(joinedload('volume_metadata')).\
- options(joinedload('volume_admin_metadata')).\
- options(joinedload('volume_type')).\
- filter_by(instance_uuid=instance_uuid).\
- all()
-
- if not result:
- return []
-
- return result
-
-
@require_context
def volume_get_all_by_project(context, project_id, marker, limit, sort_key,
sort_dir, filters=None):
db.volume_get_all_by_host(
self.ctxt, 'h%d' % i))
- def test_volume_get_all_by_instance_uuid(self):
- instance_uuids = []
- volumes = []
- for i in xrange(3):
- instance_uuid = str(uuidutils.uuid.uuid1())
- instance_uuids.append(instance_uuid)
- volumes.append([db.volume_create(self.ctxt,
- {'instance_uuid': instance_uuid})
- for j in xrange(3)])
- for i in xrange(3):
- self._assertEqualListsOfObjects(volumes[i],
- db.volume_get_all_by_instance_uuid(
- self.ctxt, instance_uuids[i]))
-
- def test_volume_get_all_by_instance_uuid_empty(self):
- self.assertEqual([], db.volume_get_all_by_instance_uuid(self.ctxt,
- 'empty'))
-
def test_volume_get_all_by_project(self):
volumes = []
for i in xrange(3):