Tests for Backup code.
"""
-import json
from xml.dom import minidom
import ddt
import mock
+from oslo_serialization import jsonutils
from oslo_utils import timeutils
import six
import webob
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual('az1', res_dict['backup']['availability_zone'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(3, len(res_dict['backups'][0]))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(2, len(res_dict['backups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(2, len(res_dict['backups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(1, len(res_dict['backups']))
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(NUM_ELEMENTS_IN_BACKUP, len(res_dict['backups'][0]))
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(1, len(res_dict['backups']))
self.assertEqual(200, res.status_int)
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(1, len(res_dict['backups']))
self.assertEqual(200, res.status_int)
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(1, len(res_dict['backups']))
self.assertEqual(200, res.status_int)
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(2, len(res_dict['backups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(2, len(res_dict['backups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(1, len(res_dict['backups']))
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['backup'])
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['backup'])
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['backup'])
self.assertTrue(_mock_service_get_all_by_topic.called)
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['backup'])
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('Invalid backup: The parent backup must be '
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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(500, res.status_int)
self.assertEqual(500, res_dict['computeFault']['code'])
self.assertEqual('Service cinder-backup could not be found.',
req = webob.Request.blank('/v2/fake/backups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('Invalid backup: No backups available to do '
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('Invalid backup: Incremental backups '
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertEqual(backup_id, res_dict['restore']['backup_id'])
req = webob.Request.blank('/v2/fake/backups/%s/restore' %
backup_id)
- req.body = json.dumps(None)
+ req.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertEqual(backup_id, res_dict['restore']['backup_id'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
description = 'auto-created_from_restore_from_backup'
# Assert that we have indeed passed on the name parameter
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertEqual(backup_id, res_dict['restore']['backup_id'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/backups/9999/restore')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(413, res.status_int)
self.assertEqual(413, res_dict['overLimit']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(413, res.status_int)
self.assertEqual(413, res_dict['overLimit']['code'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertEqual(backup_id, res_dict['restore']['backup_id'])
backup_id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertEqual(backup_id, res_dict['restore']['backup_id'])
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
# verify that request is successful
self.assertEqual(200, res.status_int)
self.assertEqual(backup_service,
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
self.assertEqual('Backup %s could not be found.' % backup_id,
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('Invalid backup: Backup status must be available '
req.headers['content-type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctx))
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
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.body = jsonutils.dump_as_bytes(body)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
# verify that request is successful
self.assertEqual(201, res.status_int)
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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
# verify that request is successful
self.assertEqual(201, res.status_int)
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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(500, res.status_int)
self.assertEqual(500, res_dict['computeFault']['code'])
self.assertEqual('Service %s could not be found.'
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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual("Invalid input received: Can't parse backup record.",
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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('Invalid backup: Backup already exists in database.',
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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(500, res.status_int)
self.assertEqual(500, res_dict['computeFault']['code'])
self.assertEqual('Service %s could not be found.' % backup_service,
# 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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('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.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
self.assertEqual('Incorrect request body format.',
ctx = context.RequestContext('admin', 'fake', is_admin=True)
req = webob.Request.blank('/v2/fake/backups/import_record')
- req.body = json.dumps(None)
+ req.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
# verify that request is successful
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertTrue(res_dict['backup']['is_incremental'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertFalse(res_dict['backup']['is_incremental'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertTrue(res_dict['backup']['is_incremental'])
Tests for consistency group code.
"""
-import ddt
-import json
from xml.dom import minidom
+import ddt
import mock
+from oslo_serialization import jsonutils
import webob
import cinder.consistencygroup
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual('az1',
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual('az1',
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(consistencygroup3.id,
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(1, len(res_dict['consistencygroups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(2, len(res_dict['consistencygroups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(2, len(res_dict['consistencygroups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(1, len(res_dict['consistencygroups']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expect_result = [consistencygroup1.id, consistencygroup2.id,
consistencygroup3.id]
expect_result.sort()
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual('az1',
req = webob.Request.blank('/v2/fake/consistencygroups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['consistencygroup'])
def test_create_consistencygroup_with_no_body(self):
# omit body from the request
req = webob.Request.blank('/v2/fake/consistencygroups')
- req.body = json.dumps(None)
+ req.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
body = {"consistencygroup": {"force": True}}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
consistencygroup = objects.ConsistencyGroup.get_by_id(
req = webob.Request.blank('/v2/fake/consistencygroups/9999/delete')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(None)
+ req.body = jsonutils.dump_as_bytes(None)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
body = {"consistencygroup": {"force": False}}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
body = {"consistencygroup": {"force": True}}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(202, res.status_int)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
body = {"invalid_request_element": {"force": False}}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
body = {"consistencygroup": {"force": "abcd"}}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
body = {"consistencygroup": {"force": ""}}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
req = webob.Request.blank('/v2/fake/consistencygroups')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
"description": description,
"add_volumes": add_volumes,
"remove_volumes": remove_volumes, }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
consistencygroup = objects.ConsistencyGroup.get_by_id(
"description": None,
"add_volumes": "fake-volume-uuid",
"remove_volumes": None, }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
"description": "new description",
"add_volumes": None,
"remove_volumes": "fake-volume-uuid", }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
"description": "",
"add_volumes": None,
"remove_volumes": None, }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
"description": "",
"add_volumes": add_volumes,
"remove_volumes": None, }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
"description": "",
"add_volumes": add_volumes,
"remove_volumes": None, }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
"description": "",
"add_volumes": add_volumes,
"remove_volumes": None, }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
"description": None,
"add_volumes": None,
"remove_volumes": None, }}
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['consistencygroup'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['consistencygroup'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/consistencygroups/create_from_src')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
import datetime
import iso8601
-import json
import uuid
import mock
req = webob.Request.blank('/v2/fake/volumes/%s/action' %
self.UUID)
req.method = 'POST'
- req.body = jsonutils.dumps({_action: None})
+ req.body = jsonutils.dump_as_bytes({_action: None})
req.content_type = 'application/json'
res = req.get_response(app)
self.assertEqual(202, res.status_int)
body = {'os-initialize_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-initialize_connection': {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-initialize_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-terminate_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-terminate_connection': {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-terminate_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
'mode': 'rw'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
req.headers["content-type"] = "application/json"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(202, res.status_int)
'mountpoint': '/dev/vdc'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-detach': {'attachment_id': 'fakeuuid'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
req.headers["content-type"] = "application/json"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
req.headers["content-type"] = "application/json"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
body = {'os-attach': {'host_name': 'fake_host',
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
req.headers["content-type"] = "application/json"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
body = {'os-begin_detaching': {'fake': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-roll_detaching': {'fake': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-extend': {'new_size': 5}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {'os-extend': {'new_size': 5}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
body = {"os-update_readonly_flag": {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(return_code, res.status_int)
body = {"os-set_bootable": {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(return_code, res.status_int)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
retype_body = {'new_type': new_type, 'migration_policy': 'never'}
- req.body = jsonutils.dumps({'os-retype': retype_body})
+ req.body = jsonutils.dump_as_bytes({'os-retype': retype_body})
res = req.get_response(fakes.wsgi_app())
self.assertEqual(expected_status, res.status_int)
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = 'POST'
req.headers['content-type'] = 'application/json'
- req.body = jsonutils.dumps({'os-retype': None})
+ req.body = jsonutils.dump_as_bytes({'os-retype': None})
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
retype_body = {'new_type': 'foo', 'migration_policy': 'invalid'}
- req.body = jsonutils.dumps({'os-retype': retype_body})
+ req.body = jsonutils.dump_as_bytes({'os-retype': retype_body})
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
req = webob.Request.blank('/v2/tenant1/volumes/%s/action' % id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
req = webob.Request.blank('/v2/tenant1/volumes/%s/action' % id)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
# License for the specific language governing permissions and limitations
# under the License.
-import json
import uuid
from xml.dom import minidom
return res
def _get_image_metadata(self, body):
- return json.loads(body)['volume']['volume_image_metadata']
+ return jsonutils.loads(body)['volume']['volume_image_metadata']
def _get_image_metadata_list(self, body):
return [
volume['volume_image_metadata']
- for volume in json.loads(body)['volumes']
+ for volume in jsonutils.loads(body)['volumes']
]
def _create_volume_and_glance_metadata(self):
body = {"os-set_image_metadata": {"metadata": fake_image_metadata}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
- json.loads(res.body)["metadata"])
+ jsonutils.loads(res.body)["metadata"])
def test_create_with_keys_case_insensitive(self):
# If the keys in uppercase_and_lowercase, should return the one
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = 'POST'
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
- json.loads(res.body)["metadata"])
+ jsonutils.loads(res.body)["metadata"])
def test_create_empty_body(self):
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
body = {"os-set_image_metadata": {
"metadata": {"image_name": "fake"}}
}
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, req, 1, body)
}
# Test for long key
- req.body = jsonutils.dumps(data)
+ req.body = jsonutils.dump_as_bytes(data)
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.create, req, 1, data)
data = {"os-set_image_metadata": {
"metadata": {"key": "v" * 260}}
}
- req.body = jsonutils.dumps(data)
+ req.body = jsonutils.dump_as_bytes(data)
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.create, req, 1, data)
data = {"os-set_image_metadata": {
"metadata": {"": "value1"}}
}
- req.body = jsonutils.dumps(data)
+ req.body = jsonutils.dump_as_bytes(data)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create, req, 1, data)
}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = 'POST'
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
}
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
req.method = 'POST'
- req.body = jsonutils.dumps(data)
+ req.body = jsonutils.dump_as_bytes(data)
req.headers["content-type"] = "application/json"
self.assertRaises(webob.exc.HTTPNotFound,
}
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
req.method = 'POST'
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
self.assertRaises(webob.exc.HTTPNotFound,
body = {"os-show_image_metadata": None}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = 'POST'
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
- json.loads(res.body)["metadata"])
+ jsonutils.loads(res.body)["metadata"])
class ImageMetadataXMLDeserializer(common.MetadataXMLDeserializer):
Tests for volume replication API code.
"""
-import json
-
import mock
from oslo_config import cfg
+from oslo_serialization import jsonutils
import webob
from cinder import context
else:
body = {'os-%s-replica' % operation: ''}
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
req.environ['cinder.context'] = context.RequestContext('admin',
'fake',
True)
Tests for volume transfer code.
"""
-import json
-import mock
from xml.dom import minidom
+import mock
+from oslo_serialization import jsonutils
import webob
from cinder.api.contrib import volume_transfer
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual('test_transfer', res_dict['transfer']['name'])
self.assertEqual(transfer['id'], res_dict['transfer']['id'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(4, len(res_dict['transfers'][0]))
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual(5, len(res_dict['transfers'][0]))
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertIn('id', res_dict['transfer'])
def test_create_transfer_with_no_body(self):
req = webob.Request.blank('/v2/fake/os-volume-transfer')
- req.body = json.dumps(None)
+ req.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
transfer['id'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
self.assertEqual(transfer['id'], res_dict['transfer']['id'])
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s/accept' %
transfer['id'])
- req.body = json.dumps(None)
+ req.body = jsonutils.dump_as_bytes(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)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
transfer['id'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
self.assertEqual(400, res_dict['badRequest']['code'])
req = webob.Request.blank('/v2/fake/os-volume-transfer/1/accept')
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(413, res.status_int)
self.assertEqual(413, res_dict['overLimit']['code'])
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
- req.body = json.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(413, res.status_int)
self.assertEqual(413, res_dict['overLimit']['code'])
# License for the specific language governing permissions and limitations
# under the License.
-import json
import mock
+from oslo_serialization import jsonutils
import webob
from cinder.api.openstack import wsgi
db.volume_type_create(context.get_admin_context(), volume_type)
return self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
def test_index(self):
res = self._get_response(volume_type)
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = stub_volume_type_encryption()
self.assertEqual(expected, res_dict)
volume_type = self._default_volume_type
res = self._get_response(volume_type)
self.assertEqual(404, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'itemNotFound': {
self._create_type_and_encryption(volume_type)
res = self._get_response(volume_type,
url='/v2/fake/types/%s/encryption/key_size')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_code)
self.assertEqual(256, res_dict['key_size'])
res = self._get_response(volume_type,
url='/v2/fake/types/%s/encryption/provider')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_code)
self.assertEqual('fake_provider', res_dict['provider'])
res = self._get_response(volume_type,
url='/v2/fake/types/%s/encryption/fake')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_code)
expected = {
self.assertEqual(0, len(self.notifier.notifications))
res = self._get_response(volume_type)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_code)
# Confirm that volume type has no encryption information
# before create.
# Create encryption specs for the volume type
# with the defined body.
res = self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(1, len(self.notifier.notifications))
# Attempt to create encryption without first creating type
res = self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(0, len(self.notifier.notifications))
self.assertEqual(404, res.status_code)
# Try to create encryption specs for a volume type
# that already has them.
res = self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'badRequest': {
# Try to create encryption specs for a volume type
# with a volume.
res = self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'badRequest': {
volume_type = self._default_volume_type
db.volume_type_create(context.get_admin_context(), volume_type)
res = self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'badRequest': {
# Test that before create, there's nothing with a get
res = self._get_response(volume_type)
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual({}, res_dict)
body = {"encryption": {'cipher': 'cipher',
# Create, and test that get returns something
res = self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
res = self._get_response(volume_type, req_method='GET',
req_headers='application/json',
url='/v2/fake/types/%s/encryption')
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(volume_type['id'], res_dict['volume_type_id'])
# Delete, and test that get returns nothing
req_headers='application/json',
url='/v2/fake/types/%s/encryption')
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual({}, res_dict)
db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
# Create encryption with volume type, and test with GET
res = self._get_response(volume_type, req_method='POST',
- req_body=json.dumps(body),
+ req_body=jsonutils.dump_as_bytes(body),
req_headers='application/json')
res = self._get_response(volume_type, req_method='GET',
req_headers='application/json',
url='/v2/fake/types/%s/encryption')
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(volume_type['id'], res_dict['volume_type_id'])
# Create volumes with the volume type
req_headers='application/json',
url='/v2/fake/types/%s/encryption/provider')
self.assertEqual(400, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'badRequest': {
'code': 400,
req_headers='application/json',
url='/v2/fake/types/%s/encryption')
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual({}, res_dict)
db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
"code": 404
}
}
- self.assertEqual(expected, json.loads(res.body))
+ self.assertEqual(expected, jsonutils.loads(res.body))
db.volume_type_destroy(context.get_admin_context(), volume_type['id'])
@mock.patch('cinder.api.openstack.wsgi.Controller.validate_integer')
res = self.\
_get_response(volume_type, req_method='PUT',
- req_body=json.dumps(update_body),
+ req_body=jsonutils.dump_as_bytes(update_body),
req_headers='application/json',
url='/v2/fake/types/%s/encryption/fake_type_id')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(512, res_dict['encryption']['key_size'])
self.assertEqual('fake_provider2', res_dict['encryption']['provider'])
# Get Encryption Specs
res = self._get_response(volume_type)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
# Confirm Encryption Specs
self.assertEqual(512, res_dict['key_size'])
# Update Encryption
res = self.\
_get_response(volume_type, req_method='PUT',
- req_body=json.dumps(update_body),
+ req_body=jsonutils.dump_as_bytes(update_body),
req_headers='application/json',
url='/v2/fake/types/%s/encryption/fake_type_id')
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'badRequest': {
# Get the Encryption
res = self._get_response(volume_type)
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
self.assertEqual(volume_type['id'], res_dict['volume_type_id'])
# Update, and test that there is an error since volumes exist
res = self.\
_get_response(volume_type, req_method='PUT',
- req_body=json.dumps(update_body),
+ req_body=jsonutils.dump_as_bytes(update_body),
req_headers='application/json',
url='/v2/fake/types/%s/encryption/fake_type_id')
self.assertEqual(400, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'badRequest': {
'code': 400,
Code Review / openstack-build / cinder-build.git / commitdiff