openstack
/
python-novaclient
Code Issues Proposed changes
python-novaclient/tests/v1_0/fakes.py

469 lines
15 KiB
Python
Raw Blame History

"""
A fake server that "responds" to API methods with pre-canned responses.
All of these responses come from the spec, so if for some reason the spec's
wrong the tests might fail. I've indicated in comments the places where actual
behavior differs from the spec.
"""
from __future__ import absolute_import
import urlparse
import urllib
import httplib2
from novaclient.v1_0 import Client
from novaclient.v1_0.client import HTTPClient
from .utils import fail, assert_in, assert_not_in, assert_has_keys
def assert_equal(value_one, value_two):
assert value_one == value_two
class FakeClient(Client):
def __init__(self, username=None, password=None, project_id=None,
auth_url=None):
super(FakeClient, self).__init__('username', 'apikey',
'project_id', 'auth_url')
self.client = FakeHTTPClient()
def assert_called(self, method, url, body=None):
"""
Assert than an API method was just called.
"""
expected = (method, url)
called = self.client.callstack[-1][0:2]
assert self.client.callstack, \
"Expected %s %s but no calls were made." % expected
assert expected == called, 'Expected %s %s; got %s %s' % \
(expected + called)
if body is not None:
assert_equal(self.client.callstack[-1][2], body)
self.client.callstack = []
def assert_called_anytime(self, method, url, body=None):
"""
Assert than an API method was called anytime in the test.
"""
expected = (method, url)
assert self.client.callstack, \
"Expected %s %s but no calls were made." % expected
found = False
for entry in self.client.callstack:
called = entry[0:2]
if expected == entry[0:2]:
found = True
break
assert found, 'Expected %s; got %s' % \
(expected, self.client.callstack)
if body is not None:
assert_equal(entry[2], body)
self.client.callstack = []
def authenticate(self):
pass
class FakeHTTPClient(HTTPClient):
def __init__(self):
self.username = 'username'
self.apikey = 'apikey'
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_not_in('body', kwargs)
elif method in ['PUT', 'POST']:
assert_in('body', kwargs)
# Call the method
munged_url = url.strip('/').replace('/', '_').replace('.', '_')
callback = "%s_%s" % (method.lower(), munged_url)
if not hasattr(self, callback):
fail('Called unknown API method: %s %s' % (method, url))
# Note the call
self.callstack.append((method, url, kwargs.get('body', None)))
status, body = getattr(self, callback)(**kwargs)
return httplib2.Response({"status": status}), body
def _munge_get_url(self, url):
return url
#
# Limits
#
def get_limits(self, **kw):
return (200, {"limits": {
"rate": [
{
"verb": "POST",
"URI": "*",
"regex": ".*",
"value": 10,
"remaining": 2,
"unit": "MINUTE",
"resetTime": 1244425439
},
{
"verb": "POST",
"URI": "*/servers",
"regex": "^/servers",
"value": 50,
"remaining": 49,
"unit": "DAY", "resetTime": 1244511839
},
{
"verb": "PUT",
"URI": "*",
"regex": ".*",
"value": 10,
"remaining": 2,
"unit": "MINUTE",
"resetTime": 1244425439
},
{
"verb": "GET",
"URI": "*changes-since*",
"regex": "changes-since",
"value": 3,
"remaining": 3,
"unit": "MINUTE",
"resetTime": 1244425439
},
{
"verb": "DELETE",
"URI": "*",
"regex": ".*",
"value": 100,
"remaining": 100,
"unit": "MINUTE",
"resetTime": 1244425439
}
],
"absolute": {
"maxTotalRAMSize": 51200,
"maxIPGroups": 50,
"maxIPGroupMembers": 25
}
}})
#
# Servers
#
def get_servers(self, **kw):
return (200, {"servers": [
{'id': 1234, 'name': 'sample-server'},
{'id': 5678, 'name': 'sample-server2'}
]})
def get_servers_detail(self, **kw):
return (200, {"servers": [
{
"id": 1234,
"name": "sample-server",
"imageId": 2,
"flavorId": 1,
"hostId": "e4d909c290d0fb1ca068ffaddf22cbd0",
"status": "BUILD",
"progress": 60,
"addresses": {
"public": ["1.2.3.4", "5.6.7.8"],
"private": ["10.11.12.13"]
},
"metadata": {
"Server Label": "Web Head 1",
"Image Version": "2.1"
}
},
{
"id": 5678,
"name": "sample-server2",
"imageId": 2,
"flavorId": 1,
"hostId": "9e107d9d372bb6826bd81d3542a419d6",
"status": "ACTIVE",
"addresses": {
"public": ["9.10.11.12"],
"private": ["10.11.12.14"]
},
"metadata": {
"Server Label": "DB 1"
}
}
]})
def post_servers(self, body, **kw):
assert_equal(body.keys(), ['server'])
assert_has_keys(body['server'],
required=['name', 'imageId', 'flavorId'],
optional=['sharedIpGroupId', 'metadata',
'personality', 'min_count', 'max_count'])
if 'personality' in body['server']:
for pfile in body['server']['personality']:
assert_has_keys(pfile, required=['path', 'contents'])
return (202, self.get_servers_1234()[1])
def get_servers_1234(self, **kw):
r = {'server': self.get_servers_detail()[1]['servers'][0]}
return (200, r)
def get_servers_5678(self, **kw):
r = {'server': self.get_servers_detail()[1]['servers'][1]}
return (200, r)
def put_servers_1234(self, body, **kw):
assert_equal(body.keys(), ['server'])
assert_has_keys(body['server'], optional=['name', 'adminPass'])
return (204, None)
def delete_servers_1234(self, **kw):
return (202, None)
#
# Server Addresses
#
def get_servers_1234_ips(self, **kw):
return (200, {'addresses':
self.get_servers_1234()[1]['server']['addresses']})
def get_servers_1234_ips_public(self, **kw):
return (200, {'public':
self.get_servers_1234_ips()[1]['addresses']['public']})
def get_servers_1234_ips_private(self, **kw):
return (200, {'private':
self.get_servers_1234_ips()[1]['addresses']['private']})
def put_servers_1234_ips_public_1_2_3_4(self, body, **kw):
assert_equal(body.keys(), ['shareIp'])
assert_has_keys(body['shareIp'], required=['sharedIpGroupId',
'configureServer'])
return (202, None)
def delete_servers_1234_ips_public_1_2_3_4(self, **kw):
return (202, None)
#
# Server actions
#
def post_servers_1234_action(self, body, **kw):
assert_equal(len(body.keys()), 1)
action = body.keys()[0]
if action == 'reboot':
assert_equal(body[action].keys(), ['type'])
assert_in(body[action]['type'], ['HARD', 'SOFT'])
elif action == 'rebuild':
assert_equal(body[action].keys(), ['imageId'])
elif action == 'resize':
assert_equal(body[action].keys(), ['flavorId'])
elif action == 'confirmResize':
assert_equal(body[action], None)
# This one method returns a different response code
return (204, None)
elif action == 'revertResize':
assert_equal(body[action], None)
elif action == 'migrate':
assert_equal(body[action], None)
elif action == 'addFixedIp':
assert_equal(body[action].keys(), ['networkId'])
elif action == 'removeFixedIp':
assert_equal(body[action].keys(), ['address'])
else:
fail("Unexpected server action: %s" % action)
return (202, None)
#
# Flavors
#
def get_flavors(self, **kw):
return (200, {'flavors': [
{'id': 1, 'name': '256 MB Server'},
{'id': 2, 'name': '512 MB Server'}
]})
def get_flavors_detail(self, **kw):
return (200, {'flavors': [
{'id': 1, 'name': '256 MB Server', 'ram': 256, 'disk': 10},
{'id': 2, 'name': '512 MB Server', 'ram': 512, 'disk': 20}
]})
def get_flavors_1(self, **kw):
return (200, {'flavor': self.get_flavors_detail()[1]['flavors'][0]})
def get_flavors_2(self, **kw):
return (200, {'flavor': self.get_flavors_detail()[1]['flavors'][1]})
#
# Images
#
def get_images(self, **kw):
return (200, {'images': [
{'id': 1, 'name': 'CentOS 5.2'},
{'id': 2, 'name': 'My Server Backup'}
]})
def get_images_detail(self, **kw):
return (200, {'images': [
{
'id': 1,
'name': 'CentOS 5.2',
"updated": "2010-10-10T12:00:00Z",
"created": "2010-08-10T12:00:00Z",
"status": "ACTIVE"
},
{
"id": 743,
"name": "My Server Backup",
"serverId": 12,
"updated": "2010-10-10T12:00:00Z",
"created": "2010-08-10T12:00:00Z",
"status": "SAVING",
"progress": 80
}
]})
def get_images_1(self, **kw):
return (200, {'image': self.get_images_detail()[1]['images'][0]})
def get_images_2(self, **kw):
return (200, {'image': self.get_images_detail()[1]['images'][1]})
def post_images(self, body, **kw):
assert_equal(body.keys(), ['image'])
assert_has_keys(body['image'], required=['serverId', 'name', 'image_type', 'backup_type', 'rotation'])
return (202, self.get_images_1()[1])
def delete_images_1(self, **kw):
return (204, None)
#
# Backup schedules
#
def get_servers_1234_backup_schedule(self, **kw):
return (200, {"backupSchedule": {
"enabled": True,
"weekly": "THURSDAY",
"daily": "H_0400_0600"
}})
def post_servers_1234_backup_schedule(self, body, **kw):
assert_equal(body.keys(), ['backupSchedule'])
assert_has_keys(body['backupSchedule'], required=['enabled'],
optional=['weekly', 'daily'])
return (204, None)
def delete_servers_1234_backup_schedule(self, **kw):
return (204, None)
#
# Shared IP groups
#
def get_shared_ip_groups(self, **kw):
return (200, {'sharedIpGroups': [
{'id': 1, 'name': 'group1'},
{'id': 2, 'name': 'group2'},
]})
def get_shared_ip_groups_detail(self, **kw):
return (200, {'sharedIpGroups': [
{'id': 1, 'name': 'group1', 'servers': [1234]},
{'id': 2, 'name': 'group2', 'servers': [5678]},
]})
def get_shared_ip_groups_1(self, **kw):
return (200, {'sharedIpGroup':
self.get_shared_ip_groups_detail()[1]['sharedIpGroups'][0]})
def post_shared_ip_groups(self, body, **kw):
assert_equal(body.keys(), ['sharedIpGroup'])
assert_has_keys(body['sharedIpGroup'], required=['name'],
optional=['server'])
return (201, {'sharedIpGroup': {
'id': 10101,
'name': body['sharedIpGroup']['name'],
'servers': 'server' in body['sharedIpGroup'] and \
[body['sharedIpGroup']['server']] or None
}})
def delete_shared_ip_groups_1(self, **kw):
return (204, None)
#
# Zones
#
def get_zones(self, **kw):
return (200, {'zones': [
{'id': 1, 'api_url': 'http://foo.com', 'username': 'bob'},
{'id': 2, 'api_url': 'http://foo.com', 'username': 'alice'},
]})
def get_zones_detail(self, **kw):
return (200, {'zones': [
{'id': 1, 'api_url': 'http://foo.com', 'username': 'bob',
'password': 'qwerty'},
{'id': 2, 'api_url': 'http://foo.com', 'username': 'alice',
'password': 'password'}
]})
def get_zones_1(self, **kw):
r = {'zone': self.get_zones_detail()[1]['zones'][0]}
return (200, r)
def get_zones_2(self, **kw):
r = {'zone': self.get_zones_detail()[1]['zones'][1]}
return (200, r)
def post_zones(self, body, **kw):
assert_equal(body.keys(), ['zone'])
assert_has_keys(body['zone'],
required=['api_url', 'username', 'password'],
optional=['weight_offset', 'weight_scale'])
return (202, self.get_zones_1()[1])
def put_zones_1(self, body, **kw):
assert_equal(body.keys(), ['zone'])
assert_has_keys(body['zone'], optional=['api_url', 'username',
'password',
'weight_offset',
'weight_scale'])
return (204, None)
def delete_zones_1(self, **kw):
return (202, None)
#
# Accounts
#
def post_accounts_test_account_create_instance(self, body, **kw):
assert_equal(body.keys(), ['server'])
assert_has_keys(body['server'],
required=['name', 'imageId', 'flavorId'],
optional=['sharedIpGroupId', 'metadata',
'personality', 'min_count', 'max_count'])
if 'personality' in body['server']:
for pfile in body['server']['personality']:
assert_has_keys(pfile, required=['path', 'contents'])
return (202, self.get_servers_1234()[1])
View Git Blame Copy Permalink