'display_description="Nightly Backup" ' 'volume_id="%s" container="Container001"/>' % volume_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/xml'
req.headers['Accept'] = 'application/xml'
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
dom = minidom.parseString(res.body)
backup = dom.getElementsByTagName('backup')
self.assertTrue(backup.item(0).hasAttribute('id'))
self.assertTrue(_mock_service_get_all_by_topic.called)
db.volume_destroy(context.get_admin_context(), volume_id)
**** CubicPower OpenStack Study ****
def test_create_backup_with_no_body(self):
# omit body from the request
req = webob.Request.blank('/v2/fake/backups')
req.body = json.dumps(None)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'The server could not comply with the request since'
' it is either malformed or otherwise incorrect.')
**** CubicPower OpenStack Study ****
def test_create_backup_with_body_KeyError(self):
# omit volume_id from body
body = {"backup": {"display_name": "nightly001",
"display_description":
"Nightly Backup 03-Sep-2012",
"container": "nightlybackups",
}
}
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Incorrect request body format')
**** CubicPower OpenStack Study ****
def test_create_backup_with_VolumeNotFound(self):
body = {"backup": {"display_name": "nightly001",
"display_description":
"Nightly Backup 03-Sep-2012",
"volume_id": 9999,
"container": "nightlybackups",
}
}
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 404)
self.assertEqual(res_dict['itemNotFound']['code'], 404)
self.assertEqual(res_dict['itemNotFound']['message'],
'Volume 9999 could not be found.')
**** CubicPower OpenStack Study ****
def test_create_backup_with_InvalidVolume(self):
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=5,
status='restoring')['id']
body = {"backup": {"display_name": "nightly001",
"display_description":
"Nightly Backup 03-Sep-2012",
"volume_id": volume_id,
"container": "nightlybackups",
}
}
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid volume: Volume to be backed up must'
' be available')
@mock.patch('cinder.db.service_get_all_by_topic')
**** CubicPower OpenStack Study ****
def test_create_backup_WithOUT_enabled_backup_service(
self,
_mock_service_get_all_by_topic):
# need an enabled backup service available
_mock_service_get_all_by_topic.return_value = []
volume_id = utils.create_volume(self.context, size=2)['id']
req = webob.Request.blank('/v2/fake/backups')
body = {"backup": {"display_name": "nightly001",
"display_description":
"Nightly Backup 03-Sep-2012",
"volume_id": volume_id,
"container": "nightlybackups",
}
}
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 500)
self.assertEqual(res_dict['computeFault']['code'], 500)
self.assertEqual(res_dict['computeFault']['message'],
'Service cinder-backup could not be found.')
volume = self.volume_api.get(context.get_admin_context(), volume_id)
self.assertEqual(volume['status'], 'available')
@mock.patch('cinder.db.service_get_all_by_topic')
**** CubicPower OpenStack Study ****
def test_is_backup_service_enabled(self, _mock_service_get_all_by_topic):
test_host = 'test_host'
alt_host = 'strange_host'
empty_service = []
#service host not match with volume's host
host_not_match = [{'availability_zone': "fake_az", 'host': alt_host,
'disabled': 0, 'updated_at': timeutils.utcnow()}]
#service az not match with volume's az
az_not_match = [{'availability_zone': "strange_az", 'host': test_host,
'disabled': 0, 'updated_at': timeutils.utcnow()}]
#service disabled
disabled_service = [{'availability_zone': "fake_az",
'host': test_host,
'disabled': 1,
'updated_at': timeutils.utcnow()}]
#dead service that last reported at 20th century
dead_service = [{'availability_zone': "fake_az", 'host': alt_host,
'disabled': 0, 'updated_at': '1989-04-16 02:55:44'}]
#first service's host not match but second one works.
multi_services = [{'availability_zone': "fake_az", 'host': alt_host,
'disabled': 0, 'updated_at': timeutils.utcnow()},
{'availability_zone': "fake_az", 'host': test_host,
'disabled': 0, 'updated_at': timeutils.utcnow()}]
#Setup mock to run through the following service cases
_mock_service_get_all_by_topic.side_effect = [empty_service,
host_not_match,
az_not_match,
disabled_service,
dead_service,
multi_services]
volume_id = utils.create_volume(self.context, size=2,
host=test_host)['id']
volume = self.volume_api.get(context.get_admin_context(), volume_id)
#test empty service
self.assertEqual(self.backup_api._is_backup_service_enabled(volume,
test_host),
False)
#test host not match service
self.assertEqual(self.backup_api._is_backup_service_enabled(volume,
test_host),
False)
#test az not match service
self.assertEqual(self.backup_api._is_backup_service_enabled(volume,
test_host),
False)
#test disabled service
self.assertEqual(self.backup_api._is_backup_service_enabled(volume,
test_host),
False)
#test dead service
self.assertEqual(self.backup_api._is_backup_service_enabled(volume,
test_host),
False)
#test multi services and the last service matches
self.assertEqual(self.backup_api._is_backup_service_enabled(volume,
test_host),
True)
**** CubicPower OpenStack Study ****
def test_delete_backup_available(self):
backup_id = self._create_backup(status='available')
req = webob.Request.blank('/v2/fake/backups/%s' %
backup_id)
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
self.assertEqual(self._get_backup_attrib(backup_id, 'status'),
'deleting')
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_delete_backup_error(self):
backup_id = self._create_backup(status='error')
req = webob.Request.blank('/v2/fake/backups/%s' %
backup_id)
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
self.assertEqual(self._get_backup_attrib(backup_id, 'status'),
'deleting')
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_delete_backup_with_backup_NotFound(self):
req = webob.Request.blank('/v2/fake/backups/9999')
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 404)
self.assertEqual(res_dict['itemNotFound']['code'], 404)
self.assertEqual(res_dict['itemNotFound']['message'],
'Backup 9999 could not be found.')
**** CubicPower OpenStack Study ****
def test_delete_backup_with_InvalidBackup(self):
backup_id = self._create_backup()
req = webob.Request.blank('/v2/fake/backups/%s' %
backup_id)
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid backup: Backup status must be '
'available or error')
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_volume_id_specified_json(self):
backup_id = self._create_backup(status='available')
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=5)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 202)
self.assertEqual(res_dict['restore']['backup_id'], backup_id)
self.assertEqual(res_dict['restore']['volume_id'], volume_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_volume_id_specified_xml(self):
backup_id = self._create_backup(status='available')
volume_id = utils.create_volume(self.context, size=2)['id']
req = webob.Request.blank('/v2/fake/backups/%s/restore' % backup_id)
req.body = '' % volume_id
req.method = 'POST'
req.headers['Content-Type'] = 'application/xml'
req.headers['Accept'] = 'application/xml'
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
dom = minidom.parseString(res.body)
restore = dom.getElementsByTagName('restore')
self.assertEqual(restore.item(0).getAttribute('backup_id'),
backup_id)
self.assertEqual(restore.item(0).getAttribute('volume_id'), volume_id)
db.backup_destroy(context.get_admin_context(), backup_id)
db.volume_destroy(context.get_admin_context(), volume_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_with_no_body(self):
# omit body from the request
backup_id = self._create_backup(status='available')
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.body = json.dumps(None)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Incorrect request body format')
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_with_body_KeyError(self):
# omit restore from body
backup_id = self._create_backup(status='available')
req = webob.Request.blank('/v2/fake/backups/%s/restore' % backup_id)
body = {"": {}}
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Incorrect request body format')
@mock.patch('cinder.volume.API.create')
**** CubicPower OpenStack Study ****
def test_restore_backup_volume_id_unspecified(self,
_mock_volume_api_create):
# intercept volume creation to ensure created volume
# has status of available
def fake_volume_api_create(context, size, name, description):
volume_id = utils.create_volume(self.context, size=size)['id']
return db.volume_get(context, volume_id)
_mock_volume_api_create.side_effect = fake_volume_api_create
backup_id = self._create_backup(size=5, status='available')
body = {"restore": {}}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 202)
self.assertEqual(res_dict['restore']['backup_id'], backup_id)
@mock.patch('cinder.backup.API.restore')
**** CubicPower OpenStack Study ****
def fake_volume_api_create(context, size, name, description):
volume_id = utils.create_volume(self.context, size=size)['id']
return db.volume_get(context, volume_id)
_mock_volume_api_create.side_effect = fake_volume_api_create
backup_id = self._create_backup(size=5, status='available')
body = {"restore": {}}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 202)
self.assertEqual(res_dict['restore']['backup_id'], backup_id)
@mock.patch('cinder.backup.API.restore')
**** CubicPower OpenStack Study ****
def test_restore_backup_with_InvalidInput(self,
_mock_volume_api_restore):
msg = _("Invalid input")
_mock_volume_api_restore.side_effect = \
exception.InvalidInput(reason=msg)
backup_id = self._create_backup(status='available')
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=0)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid input received: Invalid input')
**** CubicPower OpenStack Study ****
def test_restore_backup_with_InvalidVolume(self):
backup_id = self._create_backup(status='available')
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=5,
status='attaching')['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid volume: Volume to be restored to must '
'be available')
db.volume_destroy(context.get_admin_context(), volume_id)
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_with_InvalidBackup(self):
backup_id = self._create_backup(status='restoring')
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=5)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid backup: Backup status must be available')
db.volume_destroy(context.get_admin_context(), volume_id)
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_with_BackupNotFound(self):
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=5)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/9999/restore')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 404)
self.assertEqual(res_dict['itemNotFound']['code'], 404)
self.assertEqual(res_dict['itemNotFound']['message'],
'Backup 9999 could not be found.')
db.volume_destroy(context.get_admin_context(), volume_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_with_VolumeNotFound(self):
backup_id = self._create_backup(status='available')
body = {"restore": {"volume_id": "9999", }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 404)
self.assertEqual(res_dict['itemNotFound']['code'], 404)
self.assertEqual(res_dict['itemNotFound']['message'],
'Volume 9999 could not be found.')
db.backup_destroy(context.get_admin_context(), backup_id)
@mock.patch('cinder.backup.API.restore')
**** CubicPower OpenStack Study ****
def test_restore_backup_with_VolumeSizeExceedsAvailableQuota(
self,
_mock_backup_restore):
_mock_backup_restore.side_effect = \
exception.VolumeSizeExceedsAvailableQuota(requested='2',
consumed='2',
quota='3')
backup_id = self._create_backup(status='available')
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=5)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 413)
self.assertEqual(res_dict['overLimit']['code'], 413)
self.assertEqual(res_dict['overLimit']['message'],
'Requested volume or snapshot exceeds allowed '
'Gigabytes quota. Requested 2G, quota is 3G and '
'2G has been consumed.')
@mock.patch('cinder.backup.API.restore')
**** CubicPower OpenStack Study ****
def test_restore_backup_with_VolumeLimitExceeded(self,
_mock_backup_restore):
_mock_backup_restore.side_effect = \
exception.VolumeLimitExceeded(allowed=1)
backup_id = self._create_backup(status='available')
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=5)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 413)
self.assertEqual(res_dict['overLimit']['code'], 413)
self.assertEqual(res_dict['overLimit']['message'],
'Maximum number of volumes allowed (1) exceeded')
**** CubicPower OpenStack Study ****
def test_restore_backup_to_undersized_volume(self):
backup_size = 10
backup_id = self._create_backup(status='available', size=backup_size)
# need to create the volume referenced below first
volume_size = 5
volume_id = utils.create_volume(self.context, size=volume_size)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid volume: volume size %d is too '
'small to restore backup of size %d.'
% (volume_size, backup_size))
db.volume_destroy(context.get_admin_context(), volume_id)
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_restore_backup_to_oversized_volume(self):
backup_id = self._create_backup(status='available', size=10)
# need to create the volume referenced below first
volume_id = utils.create_volume(self.context, size=15)['id']
body = {"restore": {"volume_id": volume_id, }}
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = json.dumps(body)
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 202)
self.assertEqual(res_dict['restore']['backup_id'], backup_id)
self.assertEqual(res_dict['restore']['volume_id'], volume_id)
db.volume_destroy(context.get_admin_context(), volume_id)
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_export_record_as_non_admin(self):
backup_id = self._create_backup(status='available', size=10)
req = webob.Request.blank('/v2/fake/backups/%s/export_record' %
backup_id)
req.method = 'GET'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
# request is not authorized
self.assertEqual(res.status_int, 403)
@mock.patch('cinder.backup.rpcapi.BackupAPI.export_record')
**** CubicPower OpenStack Study ****
def test_export_backup_record_id_specified_json(self,
_mock_export_record_rpc):
backup_id = self._create_backup(status='available', size=10)
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_service = 'fake'
backup_url = 'fake'
_mock_export_record_rpc.return_value = \
{'backup_service': backup_service,
'backup_url': backup_url}
req = webob.Request.blank('/v2/fake/backups/%s/export_record' %
backup_id)
req.method = 'GET'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
# verify that request is successful
self.assertEqual(res.status_int, 200)
self.assertEqual(res_dict['backup-record']['backup_service'],
backup_service)
self.assertEqual(res_dict['backup-record']['backup_url'],
backup_url)
db.backup_destroy(context.get_admin_context(), backup_id)
@mock.patch('cinder.backup.rpcapi.BackupAPI.export_record')
**** CubicPower OpenStack Study ****
def test_export_record_backup_id_specified_xml(self,
_mock_export_record_rpc):
backup_id = self._create_backup(status='available', size=10)
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_service = 'fake'
backup_url = 'fake'
_mock_export_record_rpc.return_value = \
{'backup_service': backup_service,
'backup_url': backup_url}
req = webob.Request.blank('/v2/fake/backups/%s/export_record' %
backup_id)
req.method = 'GET'
req.headers['Content-Type'] = 'application/xml'
req.headers['Accept'] = 'application/xml'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
self.assertEqual(res.status_int, 200)
dom = minidom.parseString(res.body)
export = dom.getElementsByTagName('backup-record')
self.assertEqual(export.item(0).getAttribute('backup_service'),
backup_service)
self.assertEqual(export.item(0).getAttribute('backup_url'),
backup_url)
#db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_export_record_with_bad_backup_id(self):
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_id = 'bad_id'
req = webob.Request.blank('/v2/fake/backups/%s/export_record' %
backup_id)
req.method = 'GET'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 404)
self.assertEqual(res_dict['itemNotFound']['code'], 404)
self.assertEqual(res_dict['itemNotFound']['message'],
'Backup %s could not be found.' % backup_id)
**** CubicPower OpenStack Study ****
def test_export_record_for_unavailable_backup(self):
backup_id = self._create_backup(status='restoring')
ctx = context.RequestContext('admin', 'fake', is_admin=True)
req = webob.Request.blank('/v2/fake/backups/%s/export_record' %
backup_id)
req.method = 'GET'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid backup: Backup status must be available '
'and not restoring.')
db.backup_destroy(context.get_admin_context(), backup_id)
@mock.patch('cinder.backup.rpcapi.BackupAPI.export_record')
**** CubicPower OpenStack Study ****
def test_export_record_with_unavailable_service(self,
_mock_export_record_rpc):
msg = 'fake unavailable service'
_mock_export_record_rpc.side_effect = \
exception.InvalidBackup(reason=msg)
backup_id = self._create_backup(status='available')
ctx = context.RequestContext('admin', 'fake', is_admin=True)
req = webob.Request.blank('/v2/fake/backups/%s/export_record' %
backup_id)
req.method = 'GET'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Invalid backup: %s' % msg)
db.backup_destroy(context.get_admin_context(), backup_id)
**** CubicPower OpenStack Study ****
def test_import_record_as_non_admin(self):
backup_service = 'fake'
backup_url = 'fake'
req = webob.Request.blank('/v2/fake/backups/import_record')
body = {'backup-record': {'backup_service': backup_service,
'backup_url': backup_url}}
req.body = json.dumps(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
# request is not authorized
self.assertEqual(res.status_int, 403)
@mock.patch('cinder.backup.api.API._list_backup_services')
@mock.patch('cinder.backup.rpcapi.BackupAPI.import_record')
**** CubicPower OpenStack Study ****
def test_import_record_volume_id_specified_json(self,
_mock_import_record_rpc,
_mock_list_services):
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_service = 'fake'
backup_url = 'fake'
_mock_import_record_rpc.return_value = \
{'display_name': 'fake',
'display_description': 'fake',
'container': 'fake',
'size': 1,
'service_metadata': 'fake',
'service': 'fake',
'object_count': 1,
'status': 'available',
'availability_zone': 'fake'}
_mock_list_services.return_value = ['fake']
req = webob.Request.blank('/v2/fake/backups/import_record')
body = {'backup-record': {'backup_service': backup_service,
'backup_url': backup_url}}
req.body = json.dumps(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
# verify that request is successful
self.assertEqual(res.status_int, 201)
self.assertTrue('id' in res_dict['backup'])
@mock.patch('cinder.backup.api.API._list_backup_services')
@mock.patch('cinder.backup.rpcapi.BackupAPI.import_record')
**** CubicPower OpenStack Study ****
def test_import_record_volume_id_specified_xml(self,
_mock_import_record_rpc,
_mock_list_services):
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_service = 'fake'
backup_url = 'fake'
_mock_import_record_rpc.return_value = \
{'display_name': 'fake',
'display_description': 'fake',
'container': 'fake',
'size': 1,
'service_metadata': 'fake',
'service': 'fake',
'object_count': 1,
'status': 'available',
'availability_zone': 'fake'}
_mock_list_services.return_value = ['fake']
req = webob.Request.blank('/v2/fake/backups/import_record')
req.body = (' 'backup_url="%(backup_url)s"/>') \ % {'backup_url': backup_url,
'backup_service': backup_service}
req.method = 'POST'
req.headers['Content-Type'] = 'application/xml'
req.headers['Accept'] = 'application/xml'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
self.assertEqual(res.status_int, 201)
dom = minidom.parseString(res.body)
backup = dom.getElementsByTagName('backup')
self.assertTrue(backup.item(0).hasAttribute('id'))
@mock.patch('cinder.backup.api.API._list_backup_services')
**** CubicPower OpenStack Study ****
def test_import_record_with_no_backup_services(self,
_mock_list_services):
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_service = 'fake'
backup_url = 'fake'
_mock_list_services.return_value = []
req = webob.Request.blank('/v2/fake/backups/import_record')
body = {'backup-record': {'backup_service': backup_service,
'backup_url': backup_url}}
req.body = json.dumps(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 500)
self.assertEqual(res_dict['computeFault']['code'], 500)
self.assertEqual(res_dict['computeFault']['message'],
'Service %s could not be found.'
% backup_service)
@mock.patch('cinder.backup.api.API._list_backup_services')
@mock.patch('cinder.backup.rpcapi.BackupAPI.import_record')
**** CubicPower OpenStack Study ****
def test_import_backup_with_missing_backup_services(self,
_mock_import_record,
_mock_list_services):
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_service = 'fake'
backup_url = 'fake'
_mock_list_services.return_value = ['no-match1', 'no-match2']
_mock_import_record.side_effect = \
exception.ServiceNotFound(service_id='fake')
req = webob.Request.blank('/v2/fake/backups/import_record')
body = {'backup-record': {'backup_service': backup_service,
'backup_url': backup_url}}
req.body = json.dumps(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 500)
self.assertEqual(res_dict['computeFault']['code'], 500)
self.assertEqual(res_dict['computeFault']['message'],
'Service %s could not be found.'
% backup_service)
**** CubicPower OpenStack Study ****
def test_import_record_with_missing_body_elements(self):
ctx = context.RequestContext('admin', 'fake', is_admin=True)
backup_service = 'fake'
backup_url = 'fake'
#test with no backup_service
req = webob.Request.blank('/v2/fake/backups/import_record')
body = {'backup-record': {'backup_url': backup_url}}
req.body = json.dumps(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Incorrect request body format.')
#test with no backup_url
req = webob.Request.blank('/v2/fake/backups/import_record')
body = {'backup-record': {'backup_service': backup_service}}
req.body = json.dumps(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Incorrect request body format.')
#test with no backup_url and backup_url
req = webob.Request.blank('/v2/fake/backups/import_record')
body = {'backup-record': {}}
req.body = json.dumps(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Incorrect request body format.')
**** CubicPower OpenStack Study ****
def test_import_record_with_no_body(self):
ctx = context.RequestContext('admin', 'fake', is_admin=True)
req = webob.Request.blank('/v2/fake/backups/import_record')
req.body = json.dumps(None)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
res_dict = json.loads(res.body)
# verify that request is successful
self.assertEqual(res.status_int, 400)
self.assertEqual(res_dict['badRequest']['code'], 400)
self.assertEqual(res_dict['badRequest']['message'],
'Incorrect request body format.')