Tests for cgsnapshot code.
"""
-import json
from xml.dom import minidom
import mock
+from oslo_serialization import jsonutils
import webob
from cinder.consistencygroup import api as consistencygroupAPI
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('this is a test cgsnapshot',
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(cgsnapshot1.id,
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('this is a test cgsnapshot',
req = webob.Request.blank('/v2/fake/cgsnapshots')
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['cgsnapshot'])
def test_create_cgsnapshot_with_no_body(self):
# omit body from the request
req = webob.Request.blank('/v2/fake/cgsnapshots')
- 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'])
"CG Snapshot 1",
"consistencygroup_id": consistencygroup.id}}
req = webob.Request.blank('/v2/fake/cgsnapshots')
- 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())
- 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/cgsnapshots')
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 = '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'])
body = {'id': id,
'volume_type_id': 'cedef40a-ed67-4d10-800e-17455edce175',
'volume_id': '1', }
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(self.app)
self.assertEqual(202, res.status_int)
'volume_id': '1',
'scheduler_hints': {'a': 'b'}, }
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(self.app)
self.assertEqual(202, res.status_int)
'volume_id': '1',
'scheduler_hints': 'a', }}
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(self.app)
self.assertEqual(400, res.status_int)
body = {'os-update_snapshot_status': {'status': 'available'}}
req = webob.Request.blank('/v2/fake/snapshots/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_snapshot_status': {'status': 'in-use'}}
req = webob.Request.blank('/v2/fake/snapshots/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_snapshot_status': {}}
req = webob.Request.blank('/v2/fake/snapshots/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.environ['cinder.context'] = context.RequestContext('admin',
'fake',
True)
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(app())
return res
'fake',
True)
body = {'os-unmanage': ''}
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(app())
return res
# License for the specific language governing permissions and limitations
# under the License.
-import json
-
+from oslo_serialization import jsonutils
import webob
from cinder.api.contrib import volume_encryption_metadata
% self.volume_id)
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
"encryption_key_id": "fake_key",
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(400, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {'badRequest': {'code': 400,
'message': 'Malformed request url'}}
self.assertEqual(expected, res_dict)
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(404, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {'itemNotFound': {'code': 404,
'message': 'VolumeNotFound: Volume '
'%s could not be found.'
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(200, res.status_code)
- self.assertEqual('fake_key', res.body)
+ self.assertEqual(b'fake_key', res.body)
def test_show_control(self):
req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(200, res.status_code)
- self.assertEqual('front-end', res.body)
+ self.assertEqual(b'front-end', res.body)
def test_show_provider(self):
req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(200, res.status_code)
- self.assertEqual('nova.volume.encryptors.base.VolumeEncryptor',
+ self.assertEqual(b'nova.volume.encryptors.base.VolumeEncryptor',
res.body)
def test_show_bad_tenant_id(self):
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(400, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {'badRequest': {'code': 400,
'message': 'Malformed request url'}}
self.assertEqual(expected, res_dict)
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(404, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {'itemNotFound': {'code': 404,
'message': 'VolumeNotFound: Volume '
'%s could not be found.'
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctxt))
self.assertEqual(200, res.status_code)
- self.assertEqual('fake_key', res.body)
+ self.assertEqual(b'fake_key', res.body)
def test_show_volume_not_encrypted_type(self):
self.stubs.Set(db.sqlalchemy.api, 'volume_type_encryption_get',
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.ctxt))
self.assertEqual(200, res.status_code)
- res_dict = json.loads(res.body)
+ res_dict = jsonutils.loads(res.body)
expected = {
'encryption_key_id': None
# License for the specific language governing permissions and limitations
# under the License.
-import json
import uuid
from lxml import etree
+from oslo_serialization import jsonutils
from oslo_utils import timeutils
import webob
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volume']
+ vol = jsonutils.loads(res.body)['volume']
self.assertEqual('host001', vol['os-vol-host-attr:host'])
def test_get_volume_unallowed(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volume']
+ vol = jsonutils.loads(res.body)['volume']
self.assertNotIn('os-vol-host-attr:host', vol)
def test_list_detail_volumes_allowed(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertEqual('host001', vol[0]['os-vol-host-attr:host'])
def test_list_detail_volumes_unallowed(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertNotIn('os-vol-host-attr:host', vol[0])
def test_list_simple_volumes_no_host(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertNotIn('os-vol-host-attr:host', vol[0])
def test_get_volume_xml(self):
req.environ['cinder.context'] = context.RequestContext('admin',
'fake',
True)
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(app())
return res
# License for the specific language governing permissions and limitations
# under the License.
-import json
import uuid
from lxml import etree
+from oslo_serialization import jsonutils
from oslo_utils import timeutils
import webob
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volume']
+ vol = jsonutils.loads(res.body)['volume']
self.assertEqual('migrating', vol['os-vol-mig-status-attr:migstat'])
self.assertEqual('fake2', vol['os-vol-mig-status-attr:name_id'])
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volume']
+ vol = jsonutils.loads(res.body)['volume']
self.assertNotIn('os-vol-mig-status-attr:migstat', vol)
self.assertNotIn('os-vol-mig-status-attr:name_id', vol)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertEqual('migrating', vol[0]['os-vol-mig-status-attr:migstat'])
self.assertEqual('fake2', vol[0]['os-vol-mig-status-attr:name_id'])
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertNotIn('os-vol-mig-status-attr:migstat', vol[0])
self.assertNotIn('os-vol-mig-status-attr:name_id', vol[0])
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertNotIn('os-vol-mig-status-attr:migstat', vol[0])
self.assertNotIn('os-vol-mig-status-attr:name_id', vol[0])
# License for the specific language governing permissions and limitations
# under the License.
-import json
import uuid
from lxml import etree
+from oslo_serialization import jsonutils
import webob
from cinder import context
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volume']
+ vol = jsonutils.loads(res.body)['volume']
self.assertEqual(PROJECT_ID, vol['os-vol-tenant-attr:tenant_id'])
def test_get_volume_unallowed(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volume']
+ vol = jsonutils.loads(res.body)['volume']
self.assertNotIn('os-vol-tenant-attr:tenant_id', vol)
def test_list_detail_volumes_allowed(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertEqual(PROJECT_ID, vol[0]['os-vol-tenant-attr:tenant_id'])
def test_list_detail_volumes_unallowed(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertNotIn('os-vol-tenant-attr:tenant_id', vol[0])
def test_list_simple_volumes_no_tenant_id(self):
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
- vol = json.loads(res.body)['volumes']
+ vol = jsonutils.loads(res.body)['volumes']
self.assertNotIn('os-vol-tenant-attr:tenant_id', vol[0])
def test_get_volume_xml(self):
'fake',
True)
body = {'os-unmanage': ''}
- req.body = jsonutils.dumps(body)
+ req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(app())
return res
def check_volume_filters(self, filters):
booleans = self.db.get_booleans_for_table('volume')
- for k, v in filters.iteritems():
+ for k, v in filters.items():
try:
if k in booleans:
filters[k] = bool(v)
cinder.tests.unit.api.contrib.test_admin_actions
cinder.tests.unit.api.contrib.test_availability_zones
cinder.tests.unit.api.contrib.test_capabilities
+cinder.tests.unit.api.contrib.test_cgsnapshots
cinder.tests.unit.api.contrib.test_extended_snapshot_attributes
cinder.tests.unit.api.contrib.test_hosts
cinder.tests.unit.api.contrib.test_qos_specs_manage
cinder.tests.unit.api.contrib.test_quotas
cinder.tests.unit.api.contrib.test_quotas_classes
+cinder.tests.unit.api.contrib.test_scheduler_hints
cinder.tests.unit.api.contrib.test_scheduler_stats
cinder.tests.unit.api.contrib.test_services
+cinder.tests.unit.api.contrib.test_snapshot_actions
+cinder.tests.unit.api.contrib.test_snapshot_manage
+cinder.tests.unit.api.contrib.test_snapshot_unmanage
cinder.tests.unit.api.contrib.test_types_extra_specs
cinder.tests.unit.api.contrib.test_types_manage
cinder.tests.unit.api.contrib.test_used_limits
+cinder.tests.unit.api.contrib.test_volume_encryption_metadata
+cinder.tests.unit.api.contrib.test_volume_host_attribute
+cinder.tests.unit.api.contrib.test_volume_manage
+cinder.tests.unit.api.contrib.test_volume_unmanage
+cinder.tests.unit.api.contrib.test_volume_migration_status_attribute
+cinder.tests.unit.api.contrib.test_volume_tenant_attribute
cinder.tests.unit.api.openstack.test_wsgi
cinder.tests.unit.api.test_common
cinder.tests.unit.api.test_extensions
cinder.tests.unit.api.test_versions
cinder.tests.unit.api.test_xmlutil
+cinder.tests.unit.api.v2.test_volumes
cinder.tests.unit.image.test_cache
cinder.tests.unit.image.test_glance
cinder.tests.unit.keymgr.test_barbican