from cinder import exception
from cinder.openstack.common import jsonutils
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
+from cinder.tests.api.v2 import stubs
from cinder.volume import api as volume_api
# no auth, just let environ['cinder.context'] pass through
api = fakes.router.APIRouter()
mapper = fakes.urlmap.URLMap()
- mapper['/v1'] = api
+ mapper['/v2'] = api
return mapper
ctx = context.RequestContext('admin', 'fake', True)
# current status is available
volume = db.volume_create(ctx, {'status': 'available'})
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# request status of 'error'
# current status is 'error'
volume = db.volume_create(context.get_admin_context(),
{'status': 'error'})
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# request changing status to available
ctx = context.RequestContext('admin', 'fake', True)
# current status is available
volume = db.volume_create(ctx, {'status': 'available'})
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# malformed request body
ctx = context.RequestContext('admin', 'fake', True)
# current status is available
volume = db.volume_create(ctx, {'status': 'available'})
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# 'invalid' is not a valid status
# admin context
ctx = context.RequestContext('admin', 'fake', True)
# missing-volume-id
- req = webob.Request.blank('/v1/fake/volumes/%s/action' %
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' %
'missing-volume-id')
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# current status is available
volume = db.volume_create(ctx, {'status': 'available',
'attach_status': 'attached'})
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# request update attach_status to detached
# current status is available
volume = db.volume_create(ctx, {'status': 'available',
'attach_status': 'detached'})
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# 'invalid' is not a valid attach_status
volume = db.volume_create(ctx, {})
snapshot = db.snapshot_create(ctx, {'status': 'error_deleting',
'volume_id': volume['id']})
- req = webob.Request.blank('/v1/fake/snapshots/%s/action' %
+ req = webob.Request.blank('/v2/fake/snapshots/%s/action' %
snapshot['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
volume = db.volume_create(ctx, {})
snapshot = db.snapshot_create(ctx, {'status': 'available',
'volume_id': volume['id']})
- req = webob.Request.blank('/v1/fake/snapshots/%s/action' %
+ req = webob.Request.blank('/v2/fake/snapshots/%s/action' %
snapshot['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
ctx = context.RequestContext('admin', 'fake', True)
# current status is creating
volume = db.volume_create(ctx, {'status': 'creating'})
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
req.body = jsonutils.dumps({'os-force_delete': {}})
snapshot = db.snapshot_create(ctx, {'status': 'creating',
'volume_size': 1,
'volume_id': volume['id']})
- path = '/v1/fake/snapshots/%s/action' % snapshot['id']
+ path = '/v2/fake/snapshots/%s/action' % snapshot['id']
req = webob.Request.blank(path)
req.method = 'POST'
req.headers['content-type'] = 'application/json'
self.volume_api.reserve_volume(ctx, volume)
self.volume_api.initialize_connection(ctx, volume, {})
mountpoint = '/dev/vbd'
- self.volume_api.attach(ctx, volume, fakes.FAKE_UUID, mountpoint)
+ self.volume_api.attach(ctx, volume, stubs.FAKE_UUID, mountpoint)
# volume is attached
volume = db.volume_get(ctx, volume['id'])
self.assertEquals(volume['status'], 'in-use')
- self.assertEquals(volume['instance_uuid'], fakes.FAKE_UUID)
+ self.assertEquals(volume['instance_uuid'], stubs.FAKE_UUID)
self.assertEquals(volume['mountpoint'], mountpoint)
self.assertEquals(volume['attach_status'], 'attached')
# build request to force detach
- req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])
req.method = 'POST'
req.headers['content-type'] = 'application/json'
# request status of 'error'
from cinder import flags
from cinder.openstack.common import jsonutils
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
from cinder import volume
self.assertEqual(snapshot.get('%sprogress' % self.prefix), progress)
def test_show(self):
- url = '/v1/fake/snapshots/%s' % UUID2
+ url = '/v2/fake/snapshots/%s' % UUID2
res = self._make_request(url)
self.assertEqual(res.status_int, 200)
progress='0%')
def test_detail(self):
- url = '/v1/fake/snapshots/detail'
+ url = '/v2/fake/snapshots/detail'
res = self._make_request(url)
self.assertEqual(res.status_int, 200)
raise exception.InstanceNotFound()
self.stubs.Set(volume.api.API, 'get_snapshot', fake_snapshot_get)
- url = '/v1/fake/snapshots/70f6db34-de8d-4fbd-aafb-4065bdfa6115'
+ url = '/v2/fake/snapshots/70f6db34-de8d-4fbd-aafb-4065bdfa6115'
res = self._make_request(url)
self.assertEqual(res.status_int, 404)
from cinder.api.contrib import types_extra_specs
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
import cinder.wsgi
def setUp(self):
super(VolumeTypesExtraSpecsTest, self).setUp()
self.stubs.Set(cinder.db, 'volume_type_get', volume_type_get)
- self.api_path = '/v1/fake/os-volume-types/1/extra_specs'
+ self.api_path = '/v2/fake/os-volume-types/1/extra_specs'
self.controller = types_extra_specs.VolumeTypeExtraSpecsController()
def test_index(self):
from cinder.api.contrib import types_manage
from cinder import exception
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
from cinder.volume import volume_types
self.stubs.Set(volume_types, 'destroy',
return_volume_types_destroy)
- req = fakes.HTTPRequest.blank('/v1/fake/types/1')
+ req = fakes.HTTPRequest.blank('/v2/fake/types/1')
self.controller._delete(req, 1)
def test_volume_types_delete_not_found(self):
self.stubs.Set(volume_types, 'destroy',
return_volume_types_destroy)
- req = fakes.HTTPRequest.blank('/v1/fake/types/777')
+ req = fakes.HTTPRequest.blank('/v2/fake/types/777')
self.assertRaises(webob.exc.HTTPNotFound, self.controller._delete,
req, '777')
body = {"volume_type": {"name": "vol_type_1",
"extra_specs": {"key1": "value1"}}}
- req = fakes.HTTPRequest.blank('/v1/fake/types')
+ req = fakes.HTTPRequest.blank('/v2/fake/types')
res_dict = self.controller._create(req, body)
self.assertEqual(1, len(res_dict))
from cinder.openstack.common import jsonutils
from cinder.openstack.common.rpc import common as rpc_common
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
+from cinder.tests.api.v2 import stubs
from cinder import volume
from cinder.volume import api as volume_api
def test_simple_api_actions(self):
app = fakes.wsgi_app()
for _action in self._actions:
- req = webob.Request.blank('/v1/fake/volumes/%s/action' %
+ req = webob.Request.blank('/v2/fake/volumes/%s/action' %
self.UUID)
req.method = 'POST'
req.body = jsonutils.dumps({_action: None})
fake_initialize_connection)
body = {'os-initialize_connection': {'connector': 'fake'}}
- req = webob.Request.blank('/v1/fake/volumes/1/action')
+ req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
req.body = jsonutils.dumps(body)
req.headers["content-type"] = "application/json"
fake_terminate_connection)
body = {'os-terminate_connection': {'connector': 'fake'}}
- req = webob.Request.blank('/v1/fake/volumes/1/action')
+ req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
req.body = jsonutils.dumps(body)
req.headers["content-type"] = "application/json"
def test_attach(self):
body = {'os-attach': {'instance_uuid': 'fake',
'mountpoint': '/dev/vdc'}}
- req = webob.Request.blank('/v1/fake/volumes/1/action')
+ req = webob.Request.blank('/v2/fake/volumes/1/action')
req.method = "POST"
req.body = jsonutils.dumps(body)
req.headers["content-type"] = "application/json"
def stub_volume_get(self, context, volume_id):
- volume = fakes.stub_volume(volume_id)
+ volume = stubs.stub_volume(volume_id)
if volume_id == 5:
volume['status'] = 'in-use'
else:
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
- req = fakes.HTTPRequest.blank('/v1/tenant1/volumes/%s/action' % id)
+ req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
res_dict = self.controller._volume_upload_image(req, id, body)
expected = {'os-volume_upload_image': {'id': id,
'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1),
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
- req = fakes.HTTPRequest.blank('/v1/tenant1/volumes/%s/action' % id)
+ req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
self.assertRaises(webob.exc.HTTPNotFound,
self.controller._volume_upload_image,
req,
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
- req = fakes.HTTPRequest.blank('/v1/tenant1/volumes/%s/action' % id)
+ req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._volume_upload_image,
req,
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
- req = fakes.HTTPRequest.blank('/v1/tenant1/volumes/%s/action' % id)
+ req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._volume_upload_image,
req,
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
- req = fakes.HTTPRequest.blank('/v1/tenant1/volumes/%s/action' % id)
+ req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._volume_upload_image,
req,
from cinder import context
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
from cinder import volume
# no auth, just let environ['cinder.context'] pass through
api = fakes.router.APIRouter()
mapper = fakes.urlmap.URLMap()
- mapper['/v1'] = api
+ mapper['/v2'] = api
return mapper
def test_get_volume_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/%s' % self.UUID)
+ req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_get_volume_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
- req = webob.Request.blank('/v1/fake/volumes/%s' % self.UUID)
+ req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_list_detail_volumes_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/detail')
+ req = webob.Request.blank('/v2/fake/volumes/detail')
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_list_detail_volumes_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
- req = webob.Request.blank('/v1/fake/volumes/detail')
+ req = webob.Request.blank('/v2/fake/volumes/detail')
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_list_simple_volumes_no_host(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes')
+ req = webob.Request.blank('/v2/fake/volumes')
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_get_volume_xml(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/%s' % self.UUID)
+ req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
req.method = 'GET'
req.accept = 'application/xml'
req.environ['cinder.context'] = ctx
def test_list_volumes_detail_xml(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/detail')
+ req = webob.Request.blank('/v2/fake/volumes/detail')
req.method = 'GET'
req.accept = 'application/xml'
req.environ['cinder.context'] = ctx
from cinder import context
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
from cinder import volume
# no auth, just let environ['cinder.context'] pass through
api = fakes.router.APIRouter()
mapper = fakes.urlmap.URLMap()
- mapper['/v1'] = api
+ mapper['/v2'] = api
return mapper
def test_get_volume_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/%s' % self.UUID)
+ req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_get_volume_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
- req = webob.Request.blank('/v1/fake/volumes/%s' % self.UUID)
+ req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_list_detail_volumes_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/detail')
+ req = webob.Request.blank('/v2/fake/volumes/detail')
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_list_detail_volumes_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
- req = webob.Request.blank('/v1/fake/volumes/detail')
+ req = webob.Request.blank('/v2/fake/volumes/detail')
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_list_simple_volumes_no_tenant_id(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes')
+ req = webob.Request.blank('/v2/fake/volumes')
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
def test_get_volume_xml(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/%s' % self.UUID)
+ req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
req.method = 'GET'
req.accept = 'application/xml'
req.environ['cinder.context'] = ctx
def test_list_volumes_detail_xml(self):
ctx = context.RequestContext('admin', 'fake', True)
- req = webob.Request.blank('/v1/fake/volumes/detail')
+ req = webob.Request.blank('/v2/fake/volumes/detail')
req.method = 'GET'
req.accept = 'application/xml'
req.environ['cinder.context'] = ctx
# License for the specific language governing permissions and limitations
# under the License.
-import datetime
import uuid
import routes
from cinder.api.middleware import fault
from cinder.api.openstack import wsgi as os_wsgi
from cinder.api import urlmap
-from cinder.api.v1 import limits
-from cinder.api.v1 import router
+from cinder.api.v2 import limits
+from cinder.api.v2 import router
from cinder.api import versions
from cinder import context
-from cinder import exception as exc
from cinder.openstack.common import timeutils
from cinder import wsgi
return self.application
-def wsgi_app(inner_app_v1=None, fake_auth=True, fake_auth_context=None,
+def wsgi_app(inner_app_v2=None, fake_auth=True, fake_auth_context=None,
use_no_auth=False, ext_mgr=None):
- if not inner_app_v1:
- inner_app_v1 = router.APIRouter(ext_mgr)
+ if not inner_app_v2:
+ inner_app_v2 = router.APIRouter(ext_mgr)
if fake_auth:
if fake_auth_context is not None:
ctxt = fake_auth_context
else:
ctxt = context.RequestContext('fake', 'fake', auth_token=True)
- api_v1 = fault.FaultWrapper(auth.InjectContext(ctxt,
- inner_app_v1))
+ api_v2 = fault.FaultWrapper(auth.InjectContext(ctxt,
+ inner_app_v2))
elif use_no_auth:
- api_v1 = fault.FaultWrapper(auth.NoAuthMiddleware(
- limits.RateLimitingMiddleware(inner_app_v1)))
+ api_v2 = fault.FaultWrapper(auth.NoAuthMiddleware(
+ limits.RateLimitingMiddleware(inner_app_v2)))
else:
- api_v1 = fault.FaultWrapper(auth.AuthMiddleware(
- limits.RateLimitingMiddleware(inner_app_v1)))
+ api_v2 = fault.FaultWrapper(auth.AuthMiddleware(
+ limits.RateLimitingMiddleware(inner_app_v2)))
mapper = urlmap.URLMap()
- mapper['/v1'] = api_v1
+ mapper['/v2'] = api_v2
mapper['/'] = fault.FaultWrapper(versions.Versions())
return mapper
if not token in FAKE_UUIDS:
FAKE_UUIDS[token] = str(uuid.uuid4())
return FAKE_UUIDS[token]
-
-
-def stub_volume(id, **kwargs):
- volume = {
- 'id': id,
- 'user_id': 'fakeuser',
- 'project_id': 'fakeproject',
- 'host': 'fakehost',
- 'size': 1,
- 'availability_zone': 'fakeaz',
- 'instance_uuid': 'fakeuuid',
- 'mountpoint': '/',
- 'status': 'fakestatus',
- 'attach_status': 'attached',
- 'bootable': 'false',
- 'name': 'vol name',
- 'display_name': 'displayname',
- 'display_description': 'displaydesc',
- 'created_at': datetime.datetime(1, 1, 1, 1, 1, 1),
- 'snapshot_id': None,
- 'volume_type_id': '3e196c20-3c06-11e2-81c1-0800200c9a66',
- 'volume_metadata': [],
- 'volume_type': {'name': 'vol_type_name'}}
-
- volume.update(kwargs)
- return volume
-
-
-def stub_volume_create(self, context, size, name, description, snapshot,
- **param):
- vol = stub_volume('1')
- vol['size'] = size
- vol['display_name'] = name
- vol['display_description'] = description
- try:
- vol['snapshot_id'] = snapshot['id']
- except (KeyError, TypeError):
- vol['snapshot_id'] = None
- vol['availability_zone'] = param.get('availability_zone', 'fakeaz')
- return vol
-
-
-def stub_volume_create_from_image(self, context, size, name, description,
- snapshot, volume_type, metadata,
- availability_zone):
- vol = stub_volume('1')
- vol['status'] = 'creating'
- vol['size'] = size
- vol['display_name'] = name
- vol['display_description'] = description
- vol['availability_zone'] = 'cinder'
- return vol
-
-
-def stub_volume_update(self, context, *args, **param):
- pass
-
-
-def stub_volume_delete(self, context, *args, **param):
- pass
-
-
-def stub_volume_get(self, context, volume_id):
- return stub_volume(volume_id)
-
-
-def stub_volume_get_notfound(self, context, volume_id):
- raise exc.NotFound
-
-
-def stub_volume_get_all(context, search_opts=None):
- return [stub_volume(100, project_id='fake'),
- stub_volume(101, project_id='superfake'),
- stub_volume(102, project_id='superduperfake')]
-
-
-def stub_volume_get_all_by_project(self, context, search_opts=None):
- return [stub_volume_get(self, context, '1')]
-
-
-def stub_snapshot(id, **kwargs):
- snapshot = {'id': id,
- 'volume_id': 12,
- 'status': 'available',
- 'volume_size': 100,
- 'created_at': None,
- 'display_name': 'Default name',
- 'display_description': 'Default description',
- 'project_id': 'fake'}
-
- snapshot.update(kwargs)
- return snapshot
-
-
-def stub_snapshot_get_all(self):
- return [stub_snapshot(100, project_id='fake'),
- stub_snapshot(101, project_id='superfake'),
- stub_snapshot(102, project_id='superduperfake')]
-
-
-def stub_snapshot_get_all_by_project(self, context):
- return [stub_snapshot(1)]
-
-
-def stub_snapshot_update(self, context, *args, **param):
- pass
from cinder.api.openstack import wsgi
from cinder import exception
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
class RequestTest(test.TestCase):
from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
FLAGS = flags.FLAGS
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 OpenStack LLC.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import datetime
+
+from cinder import exception as exc
+
+FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
+FAKE_UUIDS = {}
+
+
+def stub_volume(id, **kwargs):
+ volume = {
+ 'id': id,
+ 'user_id': 'fakeuser',
+ 'project_id': 'fakeproject',
+ 'host': 'fakehost',
+ 'size': 1,
+ 'availability_zone': 'fakeaz',
+ 'instance_uuid': 'fakeuuid',
+ 'mountpoint': '/',
+ 'status': 'fakestatus',
+ 'attach_status': 'attached',
+ 'bootable': 'false',
+ 'name': 'vol name',
+ 'display_name': 'displayname',
+ 'display_description': 'displaydesc',
+ 'created_at': datetime.datetime(1, 1, 1, 1, 1, 1),
+ 'snapshot_id': None,
+ 'volume_type_id': '3e196c20-3c06-11e2-81c1-0800200c9a66',
+ 'volume_metadata': [],
+ 'volume_type': {'name': 'vol_type_name'}}
+
+ volume.update(kwargs)
+ return volume
+
+
+def stub_volume_create(self, context, size, name, description, snapshot,
+ **param):
+ vol = stub_volume('1')
+ vol['size'] = size
+ vol['display_name'] = name
+ vol['display_description'] = description
+ try:
+ vol['snapshot_id'] = snapshot['id']
+ except (KeyError, TypeError):
+ vol['snapshot_id'] = None
+ vol['availability_zone'] = param.get('availability_zone', 'fakeaz')
+ return vol
+
+
+def stub_volume_create_from_image(self, context, size, name, description,
+ snapshot, volume_type, metadata,
+ availability_zone):
+ vol = stub_volume('1')
+ vol['status'] = 'creating'
+ vol['size'] = size
+ vol['display_name'] = name
+ vol['display_description'] = description
+ vol['availability_zone'] = 'cinder'
+ return vol
+
+
+def stub_volume_update(self, context, *args, **param):
+ pass
+
+
+def stub_volume_delete(self, context, *args, **param):
+ pass
+
+
+def stub_volume_get(self, context, volume_id):
+ return stub_volume(volume_id)
+
+
+def stub_volume_get_notfound(self, context, volume_id):
+ raise exc.NotFound
+
+
+def stub_volume_get_all(context, search_opts=None):
+ return [stub_volume(100, project_id='fake'),
+ stub_volume(101, project_id='superfake'),
+ stub_volume(102, project_id='superduperfake')]
+
+
+def stub_volume_get_all_by_project(self, context, search_opts=None):
+ return [stub_volume_get(self, context, '1')]
+
+
+def stub_snapshot(id, **kwargs):
+ snapshot = {'id': id,
+ 'volume_id': 12,
+ 'status': 'available',
+ 'volume_size': 100,
+ 'created_at': None,
+ 'display_name': 'Default name',
+ 'display_description': 'Default description',
+ 'project_id': 'fake'}
+
+ snapshot.update(kwargs)
+ return snapshot
+
+
+def stub_snapshot_get_all(self):
+ return [stub_snapshot(100, project_id='fake'),
+ stub_snapshot(101, project_id='superfake'),
+ stub_snapshot(102, project_id='superduperfake')]
+
+
+def stub_snapshot_get_all_by_project(self, context):
+ return [stub_snapshot(1)]
+
+
+def stub_snapshot_update(self, context, *args, **param):
+ pass
from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
+from cinder.tests.api.v1 import stubs
from cinder import volume
self.controller = snapshots.SnapshotsController()
self.stubs.Set(db, 'snapshot_get_all_by_project',
- fakes.stub_snapshot_get_all_by_project)
+ stubs.stub_snapshot_get_all_by_project)
self.stubs.Set(db, 'snapshot_get_all',
- fakes.stub_snapshot_get_all)
+ stubs.stub_snapshot_get_all)
def test_snapshot_create(self):
self.stubs.Set(volume.api.API, "create_snapshot", stub_snapshot_create)
- self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
+ self.stubs.Set(volume.api.API, 'get', stubs.stub_volume_get)
snapshot = {"volume_id": '12',
"force": False,
"display_name": "Snapshot Test Name",
self.stubs.Set(volume.api.API,
"create_snapshot_force",
stub_snapshot_create)
- self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
+ self.stubs.Set(volume.api.API, 'get', stubs.stub_volume_get)
snapshot = {"volume_id": '12',
"force": True,
"display_name": "Snapshot Test Name",
def test_snapshot_update(self):
self.stubs.Set(volume.api.API, "get_snapshot", stub_snapshot_get)
self.stubs.Set(volume.api.API, "update_snapshot",
- fakes.stub_snapshot_update)
+ stubs.stub_snapshot_update)
updates = {"display_name": "Updated Test Name", }
body = {"snapshot": updates}
req = fakes.HTTPRequest.blank('/v1/snapshots/%s' % UUID)
def test_snapshot_list_by_status(self):
def stub_snapshot_get_all_by_project(context, project_id):
return [
- fakes.stub_snapshot(1, display_name='backup1',
+ stubs.stub_snapshot(1, display_name='backup1',
status='available'),
- fakes.stub_snapshot(2, display_name='backup2',
+ stubs.stub_snapshot(2, display_name='backup2',
status='available'),
- fakes.stub_snapshot(3, display_name='backup3',
+ stubs.stub_snapshot(3, display_name='backup3',
status='creating'),
]
self.stubs.Set(db, 'snapshot_get_all_by_project',
def test_snapshot_list_by_volume(self):
def stub_snapshot_get_all_by_project(context, project_id):
return [
- fakes.stub_snapshot(1, volume_id='vol1', status='creating'),
- fakes.stub_snapshot(2, volume_id='vol1', status='available'),
- fakes.stub_snapshot(3, volume_id='vol2', status='available'),
+ stubs.stub_snapshot(1, volume_id='vol1', status='creating'),
+ stubs.stub_snapshot(2, volume_id='vol1', status='available'),
+ stubs.stub_snapshot(3, volume_id='vol2', status='available'),
]
self.stubs.Set(db, 'snapshot_get_all_by_project',
stub_snapshot_get_all_by_project)
def test_snapshot_list_by_name(self):
def stub_snapshot_get_all_by_project(context, project_id):
return [
- fakes.stub_snapshot(1, display_name='backup1'),
- fakes.stub_snapshot(2, display_name='backup2'),
- fakes.stub_snapshot(3, display_name='backup3'),
+ stubs.stub_snapshot(1, display_name='backup1'),
+ stubs.stub_snapshot(2, display_name='backup2'),
+ stubs.stub_snapshot(3, display_name='backup3'),
]
self.stubs.Set(db, 'snapshot_get_all_by_project',
stub_snapshot_get_all_by_project)
from cinder import exception
from cinder.openstack.common import timeutils
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
from cinder.volume import volume_types
from cinder import exception
from cinder import flags
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
+from cinder.tests.api.v2 import stubs
from cinder.tests.image import fake as fake_image
from cinder.volume import api as volume_api
fake_image.stub_out_image_service(self.stubs)
self.controller = volumes.VolumeController(self.ext_mgr)
- self.stubs.Set(db, 'volume_get_all', fakes.stub_volume_get_all)
+ self.stubs.Set(db, 'volume_get_all', stubs.stub_volume_get_all)
self.stubs.Set(db, 'volume_get_all_by_project',
- fakes.stub_volume_get_all_by_project)
- self.stubs.Set(volume_api.API, 'get', fakes.stub_volume_get)
- self.stubs.Set(volume_api.API, 'delete', fakes.stub_volume_delete)
+ stubs.stub_volume_get_all_by_project)
+ self.stubs.Set(volume_api.API, 'get', stubs.stub_volume_get)
+ self.stubs.Set(volume_api.API, 'delete', stubs.stub_volume_delete)
def test_volume_create(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
vol = {"size": 100,
"display_name": "Volume Test Name",
body)
def test_volume_create_with_image_id(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
test_id = "c905cedb-7281-47e4-8a62-f26bc5fc4c77"
vol = {"size": '1',
self.assertEqual(res_dict, expected)
def test_volume_create_with_image_id_and_snapshot_id(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.stubs.Set(volume_api.API, "get_snapshot", stub_snapshot_get)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
vol = {"size": '1',
body)
def test_volume_create_with_image_id_is_integer(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
vol = {"size": '1',
"display_name": "Volume Test Name",
body)
def test_volume_create_with_image_id_not_uuid_format(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
vol = {"size": '1',
"display_name": "Volume Test Name",
body)
def test_volume_update(self):
- self.stubs.Set(volume_api.API, "update", fakes.stub_volume_update)
+ self.stubs.Set(volume_api.API, "update", stubs.stub_volume_update)
updates = {
"display_name": "Updated Test Name",
}
self.assertEquals(res_dict, expected)
def test_volume_update_metadata(self):
- self.stubs.Set(volume_api.API, "update", fakes.stub_volume_update)
+ self.stubs.Set(volume_api.API, "update", stubs.stub_volume_update)
updates = {
"metadata": {"qos_max_iops": 2000}
}
req, '1', body)
def test_update_not_found(self):
- self.stubs.Set(volume_api.API, "get", fakes.stub_volume_get_notfound)
+ self.stubs.Set(volume_api.API, "get", stubs.stub_volume_get_notfound)
updates = {
"display_name": "Updated Test Name",
}
def test_volume_list(self):
self.stubs.Set(volume_api.API, 'get_all',
- fakes.stub_volume_get_all_by_project)
+ stubs.stub_volume_get_all_by_project)
req = fakes.HTTPRequest.blank('/v1/volumes')
res_dict = self.controller.index(req)
def test_volume_list_detail(self):
self.stubs.Set(volume_api.API, 'get_all',
- fakes.stub_volume_get_all_by_project)
+ stubs.stub_volume_get_all_by_project)
req = fakes.HTTPRequest.blank('/v1/volumes/detail')
res_dict = self.controller.index(req)
expected = {'volumes': [{'status': 'fakestatus',
def test_volume_list_by_name(self):
def stub_volume_get_all_by_project(context, project_id):
return [
- fakes.stub_volume(1, display_name='vol1'),
- fakes.stub_volume(2, display_name='vol2'),
- fakes.stub_volume(3, display_name='vol3'),
+ stubs.stub_volume(1, display_name='vol1'),
+ stubs.stub_volume(2, display_name='vol2'),
+ stubs.stub_volume(3, display_name='vol3'),
]
self.stubs.Set(db, 'volume_get_all_by_project',
stub_volume_get_all_by_project)
def test_volume_list_by_status(self):
def stub_volume_get_all_by_project(context, project_id):
return [
- fakes.stub_volume(1, display_name='vol1', status='available'),
- fakes.stub_volume(2, display_name='vol2', status='available'),
- fakes.stub_volume(3, display_name='vol3', status='in-use'),
+ stubs.stub_volume(1, display_name='vol1', status='available'),
+ stubs.stub_volume(2, display_name='vol2', status='available'),
+ stubs.stub_volume(3, display_name='vol3', status='in-use'),
]
self.stubs.Set(db, 'volume_get_all_by_project',
stub_volume_get_all_by_project)
def test_volume_show_no_attachments(self):
def stub_volume_get(self, context, volume_id):
- return fakes.stub_volume(volume_id, attach_status='detached')
+ return stubs.stub_volume(volume_id, attach_status='detached')
self.stubs.Set(volume_api.API, 'get', stub_volume_get)
def test_volume_show_bootable(self):
def stub_volume_get(self, context, volume_id):
- return (fakes.stub_volume(volume_id,
+ return (stubs.stub_volume(volume_id,
volume_glance_metadata=dict(foo='bar')))
self.stubs.Set(volume_api.API, 'get', stub_volume_get)
self.assertEqual(res_dict, expected)
def test_volume_show_no_volume(self):
- self.stubs.Set(volume_api.API, "get", fakes.stub_volume_get_notfound)
+ self.stubs.Set(volume_api.API, "get", stubs.stub_volume_get_notfound)
req = fakes.HTTPRequest.blank('/v1/volumes/1')
self.assertRaises(webob.exc.HTTPNotFound,
self.assertEqual(resp.status_int, 202)
def test_volume_delete_no_volume(self):
- self.stubs.Set(volume_api.API, "get", fakes.stub_volume_get_notfound)
+ self.stubs.Set(volume_api.API, "get", stubs.stub_volume_get_notfound)
req = fakes.HTTPRequest.blank('/v1/volumes/1')
self.assertRaises(webob.exc.HTTPNotFound,
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 OpenStack LLC.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import datetime
+
+from cinder import exception as exc
+
+FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
+FAKE_UUIDS = {}
+
+
+def stub_volume(id, **kwargs):
+ volume = {
+ 'id': id,
+ 'user_id': 'fakeuser',
+ 'project_id': 'fakeproject',
+ 'host': 'fakehost',
+ 'size': 1,
+ 'availability_zone': 'fakeaz',
+ 'instance_uuid': 'fakeuuid',
+ 'mountpoint': '/',
+ 'status': 'fakestatus',
+ 'attach_status': 'attached',
+ 'bootable': 'false',
+ 'name': 'vol name',
+ 'display_name': 'displayname',
+ 'display_description': 'displaydesc',
+ 'created_at': datetime.datetime(1, 1, 1, 1, 1, 1),
+ 'snapshot_id': None,
+ 'volume_type_id': '3e196c20-3c06-11e2-81c1-0800200c9a66',
+ 'volume_metadata': [],
+ 'volume_type': {'name': 'vol_type_name'}}
+
+ volume.update(kwargs)
+ return volume
+
+
+def stub_volume_create(self, context, size, name, description, snapshot,
+ **param):
+ vol = stub_volume('1')
+ vol['size'] = size
+ vol['display_name'] = name
+ vol['display_description'] = description
+ try:
+ vol['snapshot_id'] = snapshot['id']
+ except (KeyError, TypeError):
+ vol['snapshot_id'] = None
+ vol['availability_zone'] = param.get('availability_zone', 'fakeaz')
+ return vol
+
+
+def stub_volume_create_from_image(self, context, size, name, description,
+ snapshot, volume_type, metadata,
+ availability_zone):
+ vol = stub_volume('1')
+ vol['status'] = 'creating'
+ vol['size'] = size
+ vol['display_name'] = name
+ vol['display_description'] = description
+ vol['availability_zone'] = 'cinder'
+ return vol
+
+
+def stub_volume_update(self, context, *args, **param):
+ pass
+
+
+def stub_volume_delete(self, context, *args, **param):
+ pass
+
+
+def stub_volume_get(self, context, volume_id):
+ return stub_volume(volume_id)
+
+
+def stub_volume_get_notfound(self, context, volume_id):
+ raise exc.NotFound
+
+
+def stub_volume_get_all(context, search_opts=None):
+ return [stub_volume(100, project_id='fake'),
+ stub_volume(101, project_id='superfake'),
+ stub_volume(102, project_id='superduperfake')]
+
+
+def stub_volume_get_all_by_project(self, context, search_opts=None):
+ return [stub_volume_get(self, context, '1')]
+
+
+def stub_snapshot(id, **kwargs):
+ snapshot = {'id': id,
+ 'volume_id': 12,
+ 'status': 'available',
+ 'volume_size': 100,
+ 'created_at': None,
+ 'display_name': 'Default name',
+ 'display_description': 'Default description',
+ 'project_id': 'fake'}
+
+ snapshot.update(kwargs)
+ return snapshot
+
+
+def stub_snapshot_get_all(self):
+ return [stub_snapshot(100, project_id='fake'),
+ stub_snapshot(101, project_id='superfake'),
+ stub_snapshot(102, project_id='superduperfake')]
+
+
+def stub_snapshot_get_all_by_project(self, context):
+ return [stub_snapshot(1)]
+
+
+def stub_snapshot_update(self, context, *args, **param):
+ pass
from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
+from cinder.tests.api.v2 import stubs
from cinder import volume
self.controller = snapshots.SnapshotsController()
self.stubs.Set(db, 'snapshot_get_all_by_project',
- fakes.stub_snapshot_get_all_by_project)
+ stubs.stub_snapshot_get_all_by_project)
self.stubs.Set(db, 'snapshot_get_all',
- fakes.stub_snapshot_get_all)
+ stubs.stub_snapshot_get_all)
def test_snapshot_create(self):
self.stubs.Set(volume.api.API, "create_snapshot", stub_snapshot_create)
- self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
+ self.stubs.Set(volume.api.API, 'get', stubs.stub_volume_get)
snapshot = {
"volume_id": '12',
"force": False,
def test_snapshot_create_force(self):
self.stubs.Set(volume.api.API, "create_snapshot_force",
stub_snapshot_create)
- self.stubs.Set(volume.api.API, 'get', fakes.stub_volume_get)
+ self.stubs.Set(volume.api.API, 'get', stubs.stub_volume_get)
snapshot = {
"volume_id": '12',
"force": True,
def test_snapshot_update(self):
self.stubs.Set(volume.api.API, "get_snapshot", stub_snapshot_get)
self.stubs.Set(volume.api.API, "update_snapshot",
- fakes.stub_snapshot_update)
+ stubs.stub_snapshot_update)
updates = {
"display_name": "Updated Test Name",
}
def test_snapshot_list_by_status(self):
def stub_snapshot_get_all_by_project(context, project_id):
return [
- fakes.stub_snapshot(1, display_name='backup1',
+ stubs.stub_snapshot(1, display_name='backup1',
status='available'),
- fakes.stub_snapshot(2, display_name='backup2',
+ stubs.stub_snapshot(2, display_name='backup2',
status='available'),
- fakes.stub_snapshot(3, display_name='backup3',
+ stubs.stub_snapshot(3, display_name='backup3',
status='creating'),
]
self.stubs.Set(db, 'snapshot_get_all_by_project',
def test_snapshot_list_by_volume(self):
def stub_snapshot_get_all_by_project(context, project_id):
return [
- fakes.stub_snapshot(1, volume_id='vol1', status='creating'),
- fakes.stub_snapshot(2, volume_id='vol1', status='available'),
- fakes.stub_snapshot(3, volume_id='vol2', status='available'),
+ stubs.stub_snapshot(1, volume_id='vol1', status='creating'),
+ stubs.stub_snapshot(2, volume_id='vol1', status='available'),
+ stubs.stub_snapshot(3, volume_id='vol2', status='available'),
]
self.stubs.Set(db, 'snapshot_get_all_by_project',
stub_snapshot_get_all_by_project)
def test_snapshot_list_by_name(self):
def stub_snapshot_get_all_by_project(context, project_id):
return [
- fakes.stub_snapshot(1, display_name='backup1'),
- fakes.stub_snapshot(2, display_name='backup2'),
- fakes.stub_snapshot(3, display_name='backup3'),
+ stubs.stub_snapshot(1, display_name='backup1'),
+ stubs.stub_snapshot(2, display_name='backup2'),
+ stubs.stub_snapshot(3, display_name='backup3'),
]
self.stubs.Set(db, 'snapshot_get_all_by_project',
stub_snapshot_get_all_by_project)
from cinder import exception
from cinder.openstack.common import timeutils
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
from cinder.volume import volume_types
from cinder import exception
from cinder import flags
from cinder import test
-from cinder.tests.api.openstack import fakes
+from cinder.tests.api import fakes
+from cinder.tests.api.v2 import stubs
from cinder.tests.image import fake as fake_image
from cinder.volume import api as volume_api
fake_image.stub_out_image_service(self.stubs)
self.controller = volumes.VolumeController(self.ext_mgr)
- self.stubs.Set(db, 'volume_get_all', fakes.stub_volume_get_all)
+ self.stubs.Set(db, 'volume_get_all', stubs.stub_volume_get_all)
self.stubs.Set(db, 'volume_get_all_by_project',
- fakes.stub_volume_get_all_by_project)
- self.stubs.Set(volume_api.API, 'get', fakes.stub_volume_get)
- self.stubs.Set(volume_api.API, 'delete', fakes.stub_volume_delete)
+ stubs.stub_volume_get_all_by_project)
+ self.stubs.Set(volume_api.API, 'get', stubs.stub_volume_get)
+ self.stubs.Set(volume_api.API, 'delete', stubs.stub_volume_delete)
def test_volume_create(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
vol = {
"size": 100,
body)
def test_volume_create_with_image_id(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
vol = {"size": '1',
"display_name": "Volume Test Name",
self.assertEqual(res_dict, expected)
def test_volume_create_with_image_id_and_snapshot_id(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.stubs.Set(volume_api.API, "get_snapshot", stub_snapshot_get)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
vol = {
body)
def test_volume_create_with_image_id_is_integer(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
vol = {
"size": '1',
body)
def test_volume_create_with_image_id_not_uuid_format(self):
- self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
+ self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
vol = {
"size": '1',
body)
def test_volume_update(self):
- self.stubs.Set(volume_api.API, "update", fakes.stub_volume_update)
+ self.stubs.Set(volume_api.API, "update", stubs.stub_volume_update)
updates = {
"display_name": "Updated Test Name",
}
self.assertEquals(res_dict, expected)
def test_volume_update_metadata(self):
- self.stubs.Set(volume_api.API, "update", fakes.stub_volume_update)
+ self.stubs.Set(volume_api.API, "update", stubs.stub_volume_update)
updates = {
"metadata": {"qos_max_iops": 2000}
}
req, '1', body)
def test_update_not_found(self):
- self.stubs.Set(volume_api.API, "get", fakes.stub_volume_get_notfound)
+ self.stubs.Set(volume_api.API, "get", stubs.stub_volume_get_notfound)
updates = {
"display_name": "Updated Test Name",
}
def test_volume_list(self):
self.stubs.Set(volume_api.API, 'get_all',
- fakes.stub_volume_get_all_by_project)
+ stubs.stub_volume_get_all_by_project)
req = fakes.HTTPRequest.blank('/v2/volumes')
res_dict = self.controller.index(req)
def test_volume_list_detail(self):
self.stubs.Set(volume_api.API, 'get_all',
- fakes.stub_volume_get_all_by_project)
+ stubs.stub_volume_get_all_by_project)
req = fakes.HTTPRequest.blank('/v2/volumes/detail')
res_dict = self.controller.index(req)
expected = {
def test_volume_list_by_name(self):
def stub_volume_get_all_by_project(context, project_id):
return [
- fakes.stub_volume(1, display_name='vol1'),
- fakes.stub_volume(2, display_name='vol2'),
- fakes.stub_volume(3, display_name='vol3'),
+ stubs.stub_volume(1, display_name='vol1'),
+ stubs.stub_volume(2, display_name='vol2'),
+ stubs.stub_volume(3, display_name='vol3'),
]
self.stubs.Set(db, 'volume_get_all_by_project',
stub_volume_get_all_by_project)
def test_volume_list_by_status(self):
def stub_volume_get_all_by_project(context, project_id):
return [
- fakes.stub_volume(1, display_name='vol1', status='available'),
- fakes.stub_volume(2, display_name='vol2', status='available'),
- fakes.stub_volume(3, display_name='vol3', status='in-use'),
+ stubs.stub_volume(1, display_name='vol1', status='available'),
+ stubs.stub_volume(2, display_name='vol2', status='available'),
+ stubs.stub_volume(3, display_name='vol3', status='in-use'),
]
self.stubs.Set(db, 'volume_get_all_by_project',
stub_volume_get_all_by_project)
def test_volume_show_no_attachments(self):
def stub_volume_get(self, context, volume_id):
- return fakes.stub_volume(volume_id, attach_status='detached')
+ return stubs.stub_volume(volume_id, attach_status='detached')
self.stubs.Set(volume_api.API, 'get', stub_volume_get)
self.assertEqual(res_dict, expected)
def test_volume_show_no_volume(self):
- self.stubs.Set(volume_api.API, "get", fakes.stub_volume_get_notfound)
+ self.stubs.Set(volume_api.API, "get", stubs.stub_volume_get_notfound)
req = fakes.HTTPRequest.blank('/v2/volumes/1')
self.assertRaises(webob.exc.HTTPNotFound, self.controller.show,
self.assertEqual(resp.status_int, 202)
def test_volume_delete_no_volume(self):
- self.stubs.Set(volume_api.API, "get", fakes.stub_volume_get_notfound)
+ self.stubs.Set(volume_api.API, "get", stubs.stub_volume_get_notfound)
req = fakes.HTTPRequest.blank('/v2/volumes/1')
self.assertRaises(webob.exc.HTTPNotFound, self.controller.delete,
from cinder import exception
from cinder.image import glance
from cinder import test
-from cinder.tests.api.openstack import fakes
from cinder.tests.glance import stubs as glance_stubs
Code Review / openstack-build / cinder-build.git / commitdiff