b4ea550ba7
Python3 reorganized the standard library and moved several functions to different modules. Six provides a consistent interface to them through the fake six.moves module. However, the urlparse, urllib2, etc modules have been combined into one module which Six does not support so do it the old fashioned way. Change-Id: Ieb7cc7ee2a4a97807873cfe2fc3fa0a5cf3c3980 Signed-off-by: Chuck Short <chuck.short@canonical.com>
415 lines
13 KiB
Python
415 lines
13 KiB
Python
# Copyright 2013 OpenStack, LLC
|
|
#
|
|
# 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.
|
|
|
|
try:
|
|
import urlparse
|
|
except ImportError:
|
|
import urllib.parse as urlparse
|
|
|
|
from cinderclient import client as base_client
|
|
from cinderclient.tests import fakes
|
|
import cinderclient.tests.utils as utils
|
|
from cinderclient.v2 import client
|
|
|
|
|
|
def _stub_volume(**kwargs):
|
|
volume = {
|
|
'id': '1234',
|
|
'name': None,
|
|
'description': None,
|
|
"attachments": [],
|
|
"bootable": "false",
|
|
"availability_zone": "cinder",
|
|
"created_at": "2012-08-27T00:00:00.000000",
|
|
"id": '00000000-0000-0000-0000-000000000000',
|
|
"metadata": {},
|
|
"size": 1,
|
|
"snapshot_id": None,
|
|
"status": "available",
|
|
"volume_type": "None",
|
|
"links": [
|
|
{
|
|
"href": "http://localhost/v2/fake/volumes/1234",
|
|
"rel": "self"
|
|
},
|
|
{
|
|
"href": "http://localhost/fake/volumes/1234",
|
|
"rel": "bookmark"
|
|
}
|
|
],
|
|
}
|
|
volume.update(kwargs)
|
|
return volume
|
|
|
|
|
|
def _stub_snapshot(**kwargs):
|
|
snapshot = {
|
|
"created_at": "2012-08-28T16:30:31.000000",
|
|
"display_description": None,
|
|
"display_name": None,
|
|
"id": '11111111-1111-1111-1111-111111111111',
|
|
"size": 1,
|
|
"status": "available",
|
|
"volume_id": '00000000-0000-0000-0000-000000000000',
|
|
}
|
|
snapshot.update(kwargs)
|
|
return snapshot
|
|
|
|
|
|
def _self_href(base_uri, tenant_id, backup_id):
|
|
return '%s/v2/%s/backups/%s' % (base_uri, tenant_id, backup_id)
|
|
|
|
|
|
def _bookmark_href(base_uri, tenant_id, backup_id):
|
|
return '%s/%s/backups/%s' % (base_uri, tenant_id, backup_id)
|
|
|
|
|
|
def _stub_backup_full(id, base_uri, tenant_id):
|
|
return {
|
|
'id': id,
|
|
'name': 'backup',
|
|
'description': 'nightly backup',
|
|
'volume_id': '712f4980-5ac1-41e5-9383-390aa7c9f58b',
|
|
'container': 'volumebackups',
|
|
'object_count': 220,
|
|
'size': 10,
|
|
'availability_zone': 'az1',
|
|
'created_at': '2013-04-12T08:16:37.000000',
|
|
'status': 'available',
|
|
'links': [
|
|
{
|
|
'href': _self_href(base_uri, tenant_id, id),
|
|
'rel': 'self'
|
|
},
|
|
{
|
|
'href': _bookmark_href(base_uri, tenant_id, id),
|
|
'rel': 'bookmark'
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def _stub_backup(id, base_uri, tenant_id):
|
|
return {
|
|
'id': id,
|
|
'name': 'backup',
|
|
'links': [
|
|
{
|
|
'href': _self_href(base_uri, tenant_id, id),
|
|
'rel': 'self'
|
|
},
|
|
{
|
|
'href': _bookmark_href(base_uri, tenant_id, id),
|
|
'rel': 'bookmark'
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def _stub_restore():
|
|
return {'volume_id': '712f4980-5ac1-41e5-9383-390aa7c9f58b'}
|
|
|
|
|
|
class FakeClient(fakes.FakeClient, client.Client):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
client.Client.__init__(self, 'username', 'password',
|
|
'project_id', 'auth_url',
|
|
extensions=kwargs.get('extensions'))
|
|
self.client = FakeHTTPClient(**kwargs)
|
|
|
|
|
|
class FakeHTTPClient(base_client.HTTPClient):
|
|
|
|
def __init__(self, **kwargs):
|
|
self.username = 'username'
|
|
self.password = 'password'
|
|
self.auth_url = 'auth_url'
|
|
self.callstack = []
|
|
|
|
def _cs_request(self, url, method, **kwargs):
|
|
# Check that certain things are called correctly
|
|
if method in ['GET', 'DELETE']:
|
|
assert 'body' not in kwargs
|
|
elif method == 'PUT':
|
|
assert 'body' in kwargs
|
|
|
|
# Call the method
|
|
args = urlparse.parse_qsl(urlparse.urlparse(url)[4])
|
|
kwargs.update(args)
|
|
munged_url = url.rsplit('?', 1)[0]
|
|
munged_url = munged_url.strip('/').replace('/', '_').replace('.', '_')
|
|
munged_url = munged_url.replace('-', '_')
|
|
|
|
callback = "%s_%s" % (method.lower(), munged_url)
|
|
|
|
if not hasattr(self, callback):
|
|
raise AssertionError('Called unknown API method: %s %s, '
|
|
'expected fakes method name: %s' %
|
|
(method, url, callback))
|
|
|
|
# Note the call
|
|
self.callstack.append((method, url, kwargs.get('body', None)))
|
|
status, headers, body = getattr(self, callback)(**kwargs)
|
|
r = utils.TestResponse({
|
|
"status_code": status,
|
|
"text": body,
|
|
"headers": headers,
|
|
})
|
|
return r, body
|
|
|
|
if hasattr(status, 'items'):
|
|
return utils.TestResponse(status), body
|
|
else:
|
|
return utils.TestResponse({"status": status}), body
|
|
|
|
#
|
|
# Snapshots
|
|
#
|
|
|
|
def get_snapshots_detail(self, **kw):
|
|
return (200, {}, {'snapshots': [
|
|
_stub_snapshot(),
|
|
]})
|
|
|
|
def get_snapshots_1234(self, **kw):
|
|
return (200, {}, {'snapshot': _stub_snapshot(id='1234')})
|
|
|
|
def put_snapshots_1234(self, **kw):
|
|
snapshot = _stub_snapshot(id='1234')
|
|
snapshot.update(kw['body']['snapshot'])
|
|
return (200, {}, {'snapshot': snapshot})
|
|
|
|
#
|
|
# Volumes
|
|
#
|
|
|
|
def put_volumes_1234(self, **kw):
|
|
volume = _stub_volume(id='1234')
|
|
volume.update(kw['body']['volume'])
|
|
return (200, {}, {'volume': volume})
|
|
|
|
def get_volumes(self, **kw):
|
|
return (200, {}, {"volumes": [
|
|
{'id': 1234, 'name': 'sample-volume'},
|
|
{'id': 5678, 'name': 'sample-volume2'}
|
|
]})
|
|
|
|
# TODO(jdg): This will need to change
|
|
# at the very least it's not complete
|
|
def get_volumes_detail(self, **kw):
|
|
return (200, {}, {"volumes": [
|
|
{'id': 1234,
|
|
'name': 'sample-volume',
|
|
'attachments': [{'server_id': 1234}]},
|
|
]})
|
|
|
|
def get_volumes_1234(self, **kw):
|
|
r = {'volume': self.get_volumes_detail()[2]['volumes'][0]}
|
|
return (200, {}, r)
|
|
|
|
def post_volumes_1234_action(self, body, **kw):
|
|
_body = None
|
|
resp = 202
|
|
assert len(body.keys()) == 1
|
|
action = body.keys()[0]
|
|
if action == 'os-attach':
|
|
assert body[action].keys() == ['instance_uuid', 'mountpoint']
|
|
elif action == 'os-detach':
|
|
assert body[action] is None
|
|
elif action == 'os-reserve':
|
|
assert body[action] is None
|
|
elif action == 'os-unreserve':
|
|
assert body[action] is None
|
|
elif action == 'os-initialize_connection':
|
|
assert body[action].keys() == ['connector']
|
|
return (202, {}, {'connection_info': 'foos'})
|
|
elif action == 'os-terminate_connection':
|
|
assert body[action].keys() == ['connector']
|
|
elif action == 'os-begin_detaching':
|
|
assert body[action] is None
|
|
elif action == 'os-roll_detaching':
|
|
assert body[action] is None
|
|
else:
|
|
raise AssertionError("Unexpected server action: %s" % action)
|
|
return (resp, {}, _body)
|
|
|
|
def post_volumes(self, **kw):
|
|
return (202, {}, {'volume': {}})
|
|
|
|
def delete_volumes_1234(self, **kw):
|
|
return (202, {}, None)
|
|
|
|
#
|
|
# Quotas
|
|
#
|
|
|
|
def get_os_quota_sets_test(self, **kw):
|
|
return (200, {}, {'quota_set': {
|
|
'tenant_id': 'test',
|
|
'metadata_items': [],
|
|
'volumes': 1,
|
|
'snapshots': 1,
|
|
'gigabytes': 1}})
|
|
|
|
def get_os_quota_sets_test_defaults(self):
|
|
return (200, {}, {'quota_set': {
|
|
'tenant_id': 'test',
|
|
'metadata_items': [],
|
|
'volumes': 1,
|
|
'snapshots': 1,
|
|
'gigabytes': 1}})
|
|
|
|
def put_os_quota_sets_test(self, body, **kw):
|
|
assert body.keys() == ['quota_set']
|
|
fakes.assert_has_keys(body['quota_set'],
|
|
required=['tenant_id'])
|
|
return (200, {}, {'quota_set': {
|
|
'tenant_id': 'test',
|
|
'metadata_items': [],
|
|
'volumes': 2,
|
|
'snapshots': 2,
|
|
'gigabytes': 1}})
|
|
|
|
#
|
|
# Quota Classes
|
|
#
|
|
|
|
def get_os_quota_class_sets_test(self, **kw):
|
|
return (200, {}, {'quota_class_set': {
|
|
'class_name': 'test',
|
|
'metadata_items': [],
|
|
'volumes': 1,
|
|
'snapshots': 1,
|
|
'gigabytes': 1}})
|
|
|
|
def put_os_quota_class_sets_test(self, body, **kw):
|
|
assert body.keys() == ['quota_class_set']
|
|
fakes.assert_has_keys(body['quota_class_set'],
|
|
required=['class_name'])
|
|
return (200, {}, {'quota_class_set': {
|
|
'class_name': 'test',
|
|
'metadata_items': [],
|
|
'volumes': 2,
|
|
'snapshots': 2,
|
|
'gigabytes': 1}})
|
|
|
|
#
|
|
# VolumeTypes
|
|
#
|
|
def get_types(self, **kw):
|
|
return (200, {}, {
|
|
'volume_types': [{'id': 1,
|
|
'name': 'test-type-1',
|
|
'extra_specs': {}},
|
|
{'id': 2,
|
|
'name': 'test-type-2',
|
|
'extra_specs': {}}]})
|
|
|
|
def get_types_1(self, **kw):
|
|
return (200, {}, {'volume_type': {'id': 1,
|
|
'name': 'test-type-1',
|
|
'extra_specs': {}}})
|
|
|
|
def post_types(self, body, **kw):
|
|
return (202, {}, {'volume_type': {'id': 3,
|
|
'name': 'test-type-3',
|
|
'extra_specs': {}}})
|
|
|
|
def post_types_1_extra_specs(self, body, **kw):
|
|
assert body.keys() == ['extra_specs']
|
|
return (200, {}, {'extra_specs': {'k': 'v'}})
|
|
|
|
def delete_types_1_extra_specs_k(self, **kw):
|
|
return(204, {}, None)
|
|
|
|
def delete_types_1(self, **kw):
|
|
return (202, {}, None)
|
|
|
|
#
|
|
# Set/Unset metadata
|
|
#
|
|
def delete_volumes_1234_metadata_test_key(self, **kw):
|
|
return (204, {}, None)
|
|
|
|
def delete_volumes_1234_metadata_key1(self, **kw):
|
|
return (204, {}, None)
|
|
|
|
def delete_volumes_1234_metadata_key2(self, **kw):
|
|
return (204, {}, None)
|
|
|
|
def post_volumes_1234_metadata(self, **kw):
|
|
return (204, {}, {'metadata': {'test_key': 'test_value'}})
|
|
|
|
#
|
|
# List all extensions
|
|
#
|
|
def get_extensions(self, **kw):
|
|
exts = [
|
|
{
|
|
"alias": "FAKE-1",
|
|
"description": "Fake extension number 1",
|
|
"links": [],
|
|
"name": "Fake1",
|
|
"namespace": ("http://docs.openstack.org/"
|
|
"/ext/fake1/api/v1.1"),
|
|
"updated": "2011-06-09T00:00:00+00:00"
|
|
},
|
|
{
|
|
"alias": "FAKE-2",
|
|
"description": "Fake extension number 2",
|
|
"links": [],
|
|
"name": "Fake2",
|
|
"namespace": ("http://docs.openstack.org/"
|
|
"/ext/fake1/api/v1.1"),
|
|
"updated": "2011-06-09T00:00:00+00:00"
|
|
},
|
|
]
|
|
return (200, {}, {"extensions": exts, })
|
|
|
|
#
|
|
# VolumeBackups
|
|
#
|
|
|
|
def get_backups_76a17945_3c6f_435c_975b_b5685db10b62(self, **kw):
|
|
base_uri = 'http://localhost:8776'
|
|
tenant_id = '0fa851f6668144cf9cd8c8419c1646c1'
|
|
backup1 = '76a17945-3c6f-435c-975b-b5685db10b62'
|
|
return (200, {},
|
|
{'backup': _stub_backup_full(backup1, base_uri, tenant_id)})
|
|
|
|
def get_backups_detail(self, **kw):
|
|
base_uri = 'http://localhost:8776'
|
|
tenant_id = '0fa851f6668144cf9cd8c8419c1646c1'
|
|
backup1 = '76a17945-3c6f-435c-975b-b5685db10b62'
|
|
backup2 = 'd09534c6-08b8-4441-9e87-8976f3a8f699'
|
|
return (200, {},
|
|
{'backups': [
|
|
_stub_backup_full(backup1, base_uri, tenant_id),
|
|
_stub_backup_full(backup2, base_uri, tenant_id)]})
|
|
|
|
def delete_backups_76a17945_3c6f_435c_975b_b5685db10b62(self, **kw):
|
|
return (202, {}, None)
|
|
|
|
def post_backups(self, **kw):
|
|
base_uri = 'http://localhost:8776'
|
|
tenant_id = '0fa851f6668144cf9cd8c8419c1646c1'
|
|
backup1 = '76a17945-3c6f-435c-975b-b5685db10b62'
|
|
return (202, {},
|
|
{'backup': _stub_backup(backup1, base_uri, tenant_id)})
|
|
|
|
def post_backups_76a17945_3c6f_435c_975b_b5685db10b62_restore(self, **kw):
|
|
return (200, {},
|
|
{'restore': _stub_restore()})
|