commission_pending/info/action/accept/reject/resolve/issuejson/issue
- Implement resize_server and expose it as server resize [#4153]
- Expose the project API of astakosclient to kamaki CLI [#4155]
+- Full support for compute v2 basic API [#4139]
class ComputeClient(ComputeRestClient):
"""OpenStack Compute API 1.1 client"""
- def list_servers(self, detail=False):
+ def list_servers(
+ self,
+ detail=False,
+ changes_since=None,
+ image=None,
+ flavor=None,
+ name=None,
+ marker=None,
+ limit=None,
+ status=None,
+ host=None,
+ response_headers=dict(previous=None, next=None)):
"""
:param detail: if true, append full server details to each item
+ :param response_headers: (dict) use it to get previous/next responses
+ Keep the existing dict format to actually get the server responses
+ Use it with very long lists or with marker
+
:returns: list of server ids and names
"""
- detail = 'detail' if detail else ''
- r = self.servers_get(command=detail)
+ r = self.servers_get(
+ detail=bool(detail),
+ changes_since=changes_since,
+ image=image,
+ flavor=flavor,
+ name=name,
+ marker=marker,
+ limit=limit,
+ status=status,
+ host=host)
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['servers']
- def get_server_details(self, server_id, **kwargs):
+ def get_server_details(
+ self, server_id,
+ changes_since=None,
+ image=None,
+ flavor=None,
+ name=None,
+ marker=None,
+ limit=None,
+ status=None,
+ host=None,
+ response_headers=dict(previous=None, next=None),
+ **kwargs):
"""Return detailed info for a server
:param server_id: integer (int or str)
:returns: dict with server details
"""
- r = self.servers_get(server_id, **kwargs)
+ r = self.servers_get(
+ server_id,
+ changes_since=changes_since,
+ image=image,
+ flavor=flavor,
+ name=name,
+ marker=marker,
+ limit=limit,
+ status=status,
+ host=host,
+ **kwargs)
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['server']
- def create_server(self, name, flavor_id, image_id, personality=None):
+ def create_server(
+ self, name, flavor_id, image_id,
+ security_group=None,
+ user_data=None,
+ availability_zone=None,
+ metadata=None,
+ personality=None,
+ response_headers=dict(location=None)):
"""Submit request to create a new server
:param name: (str)
:param image_id: (str) id denoting the OS image to run on the VM
+ :param metadata: (dict) vm metadata
+
:param personality: a list of (file path, file contents) tuples,
describing files to be injected into VM upon creation.
:raises ClientError: wraps request errors
"""
- req = {'server': {'name': name,
- 'flavorRef': flavor_id,
- 'imageRef': image_id}}
-
- image = self.get_image_details(image_id)
- metadata = {}
- for key in ('os', 'users'):
- try:
- metadata[key] = image['metadata'][key]
- except KeyError:
- pass
+ req = {'server': {
+ 'name': name, 'flavorRef': flavor_id, 'imageRef': image_id}}
+
if metadata:
req['server']['metadata'] = metadata
if personality:
req['server']['personality'] = personality
- try:
- r = self.servers_post(json_data=req)
- except ClientError as err:
- try:
- if isinstance(err.details, list):
- tmp_err = err.details
- else:
- errd = '%s' % err.details
- tmp_err = errd.split(',')
- tmp_err = tmp_err[0].split(':')
- tmp_err = tmp_err[2].split('"')
- err.message = tmp_err[1]
- finally:
- raise err
+ r = self.servers_post(
+ json_data=req,
+ security_group=security_group,
+ user_data=user_data,
+ availability_zone=availability_zone)
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['server']
def update_server_name(self, server_id, new_name):
r = self.servers_delete(server_id)
return r.headers
+ def change_admin_password(self, server_id, new_password):
+ """
+ :param server_id: (int)
+
+ :param new_password: (str)
+ """
+ req = {"changePassword": {"adminPass": new_password}}
+ r = self.servers_action_post(server_id, json_data=req)
+ return r.headers
+
+ def rebuild_server(self, server_id, response_headers=dict(location=None)):
+ """OS"""
+ server = self.get_server_details(server_id)
+ r = self.servers_action_post(
+ server_id, json_data=dict(rebuild=server['server']))
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
+ return r.json['server']
+
def reboot_server(self, server_id, hard=False):
"""
:param server_id: integer (str or int)
:param hard: perform a hard reboot if true, soft reboot otherwise
"""
- boot_type = 'HARD' if hard else 'SOFT'
- req = {'reboot': {'type': boot_type}}
- r = self.servers_post(server_id, 'action', json_data=req)
+ req = {'reboot': {'type': 'HARD' if hard else 'SOFT'}}
+ r = self.servers_action_post(server_id, json_data=req)
return r.headers
def resize_server(self, server_id, flavor_id):
:returns: (dict) request headers
"""
req = {'resize': {'flavorRef': flavor_id}}
- r = self.servers_post(server_id, 'action', json_data=req)
+ r = self.servers_action_post(server_id, json_data=req)
+ return r.headers
+
+ def confirm_resize_server(self, server_id):
+ """OS"""
+ r = self.servers_action_post(
+ server_id, json_data=dict(confirmResize=None))
return r.headers
- def get_server_metadata(self, server_id, key=''):
+ def revert_resize_server(self, server_id):
+ """OS"""
+ r = self.servers_action_post(
+ server_id, json_data=dict(revertResize=None))
+ return r.headers
+
+ def create_server_image(self, server_id, image_name, **metadata):
+ """OpenStack method for taking snapshots"""
+ req = dict(createImage=dict(name=image_name, metadata=metadata))
+ r = self.servers_action_post(server_id, json_data=req)
+ return r.headers['location']
+
+ def start_server(self, server_id):
+ """OS Extentions"""
+ req = {'os-start': None}
+ r = self.servers_action_post(server_id, json_data=req, success=202)
+ return r.headers
+
+ def shutdown_server(self, server_id):
+ """OS Extentions"""
+ req = {'os-stop': None}
+ r = self.servers_action_post(server_id, json_data=req, success=202)
+ return r.headers
+
+ def get_server_metadata(self, server_id, key='', response_headers=dict(
+ previous=None, next=None)):
"""
:param server_id: integer (str or int)
:returns: a key:val dict of requests metadata
"""
- command = path4url('metadata', key)
- r = self.servers_get(server_id, command)
+ r = self.servers_metadata_get(server_id, key)
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['meta' if key else 'metadata']
def create_server_metadata(self, server_id, key, val):
:returns: dict of updated key:val metadata
"""
req = {'meta': {key: val}}
- r = self.servers_put(
- server_id, 'metadata/' + key, json_data=req, success=201)
+ r = self.servers_metadata_put(
+ server_id, key, json_data=req, success=201)
return r.json['meta']
- def update_server_metadata(self, server_id, **metadata):
+ def update_server_metadata(
+ self, server_id,
+ response_headers=dict(previous=None, next=None), **metadata):
"""
:param server_id: integer (str or int)
:returns: dict of updated key:val metadata
"""
req = {'metadata': metadata}
- r = self.servers_post(
- server_id, 'metadata', json_data=req, success=201)
+ r = self.servers_metadata_post(server_id, json_data=req, success=201)
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['metadata']
def delete_server_metadata(self, server_id, key):
:returns: (dict) response headers
"""
- r = self.servers_delete(server_id, 'metadata/' + key)
+ r = self.servers_metadata_delete(server_id, key)
return r.headers
- def list_flavors(self, detail=False):
+ def list_flavors(self, detail=False, response_headers=dict(
+ previous=None, next=None)):
"""
:param detail: (bool) detailed flavor info if set, short if not
:returns: (list) flavor info
"""
- r = self.flavors_get(command='detail' if detail else '')
+ r = self.flavors_get(detail=bool(detail))
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['flavors']
def get_flavor_details(self, flavor_id):
r = self.flavors_get(flavor_id)
return r.json['flavor']
- def list_images(self, detail=False):
+ def list_images(self, detail=False, response_headers=dict(
+ next=None, previous=None)):
"""
:param detail: (bool) detailed info if set, short if not
:returns: dict id,name + full info if detail
"""
- detail = 'detail' if detail else ''
- r = self.images_get(command=detail)
+ r = self.images_get(detail=bool(detail))
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['images']
def get_image_details(self, image_id, **kwargs):
r = self.images_delete(image_id)
return r.headers
- def get_image_metadata(self, image_id, key=''):
+ def get_image_metadata(self, image_id, key='', response_headers=dict(
+ previous=None, next=None)):
"""
:param image_id: (str)
:returns (dict) metadata if key not set, specific metadatum otherwise
"""
- command = path4url('metadata', key)
- r = self.images_get(image_id, command)
+ r = self.images_metadata_get(image_id, key)
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['meta' if key else 'metadata']
def create_image_metadata(self, image_id, key, val):
:returns: (dict) updated metadata
"""
req = {'meta': {key: val}}
- r = self.images_put(image_id, 'metadata/' + key, json_data=req)
+ r = self.images_metadata_put(image_id, key, json_data=req)
return r.json['meta']
- def update_image_metadata(self, image_id, **metadata):
+ def update_image_metadata(
+ self, image_id,
+ response_headers=dict(previous=None, next=None), **metadata):
"""
:param image_id: (str)
:returns: updated metadata
"""
req = {'metadata': metadata}
- r = self.images_post(image_id, 'metadata', json_data=req)
+ r = self.images_metadata_post(image_id, json_data=req)
+ for k, v in response_headers.items():
+ response_headers[k] = r.headers.get(k, v)
return r.json['metadata']
def delete_image_metadata(self, image_id, key):
:returns: (dict) response headers
"""
- command = path4url('metadata', key)
- r = self.images_delete(image_id, command)
+ r = self.images_metadata_delete(image_id, key)
return r.headers
def get_floating_ip_pools(self, tenant_id):
class ComputeRestClient(Client):
- def servers_get(self, server_id='', command='', success=200, **kwargs):
- """GET base_url/servers[/server_id][/command] request
-
- :param server_id: integer (as int or str)
-
- :param command: 'ips', 'stats', or ''
+ # NON-cyclades
+ def limits_get(self, success=200, **kwargs):
+ """GET base_url/limits
:param success: success code or list or tupple of accepted success
codes. if server response code is not in this list, a ClientError
:returns: request response
"""
- path = path4url('servers', server_id, command)
+ path = path4url('limits')
return self.get(path, success=success, **kwargs)
- def servers_delete(self, server_id='', command='', success=204, **kwargs):
- """DEL ETE base_url/servers[/server_id][/command] request
+ def servers_get(
+ self,
+ server_id='', detail=False,
+ changes_since=None,
+ image=None,
+ flavor=None,
+ name=None,
+ marker=None,
+ limit=None,
+ status=None,
+ host=None,
+ success=200,
+ **kwargs):
+ """GET base_url/servers/['detail'|<server_id>]
- :param server_id: integer (as int or str)
+ :param server_id: (int or int str) ignored if detail
- :param command: 'ips', 'stats', or ''
+ :param detail: (boolean)
- :param success: success code or list or tupple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ --- Parameters ---
+
+ :param changes-since: A time/date stamp for when the server last
+ changed status
+
+ :param image: Name of the image in URL format
+
+ :param flavor: Name of the flavor in URL format
+
+ :param name: Name of the server as a string
+
+ :param marker: UUID of the server at which you want to set a marker
+
+ :param limit: (int) limit of values to return
+
+ :param status: Status of the server (e.g. filter on "ACTIVE")
+
+ :param host: Name of the host as a string
:returns: request response
"""
- path = path4url('servers', server_id, command)
- return self.delete(path, success=success, **kwargs)
+ if not server_id:
+ self.set_param('changes-since', changes_since, iff=changes_since)
+ self.set_param('image', image, iff=image)
+ self.set_param('flavor', flavor, iff=flavor)
+ self.set_param('name', name, iff=name)
+ self.set_param('marker', marker, iff=marker)
+ self.set_param('limit', limit, iff=limit)
+ self.set_param('status', status, iff=status)
+ self.set_param('host', host, iff=host)
+
+ path = path4url('servers', 'detail' if detail else (server_id or ''))
+ return self.get(path, success=success, **kwargs)
def servers_post(
self,
- server_id='',
- command='',
+ security_group=None,
+ user_data=None,
+ availability_zone=None,
json_data=None,
success=202,
**kwargs):
- """POST base_url/servers[/server_id]/[command] request
+ """POST base_url/servers
- :param server_id: integer (as int or str)
+ :param json_data: a json-formated dict that will be send as data
- :param command: 'ips', 'stats', or ''
+ --- Parameters
- :param json_data: a json-formated dict that will be send as data
+ :param security_group: (str)
- :param success: success code or list or tupple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ :param user_data: Use to pass configuration information or scripts upon
+ launch. Must be Base64 encoded.
+
+ :param availability_zone: (str)
:returns: request response
"""
- data = json_data
+
+ self.set_param('security_group', security_group, iff=security_group)
+ self.set_param('user_data', user_data, iff=user_data)
+ self.set_param(
+ 'availability_zone', availability_zone, iff=availability_zone)
+
if json_data:
- data = json.dumps(json_data)
+ json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
- self.set_header('Content-Length', len(data))
+ self.set_header('Content-Length', len(json_data))
- path = path4url('servers', server_id, command)
- return self.post(path, data=data, success=success, **kwargs)
+ path = path4url('servers')
+ return self.post(path, data=json_data, success=success, **kwargs)
def servers_put(
self,
- server_id='',
- command='',
- json_data=None,
- success=204,
+ server_id, server_name=None, json_data=None, success=204,
**kwargs):
- """PUT base_url/servers[/server_id]/[command] request
-
- :param server_id: integer (as int or str)
-
- :param command: 'ips', 'stats', or ''
+ """PUT base_url/servers/<server_id>
:param json_data: a json-formated dict that will be send as data
- :param success: success code or list or tupple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ :param success: success code (iterable of) codes
+
+ :raises ClientError: if returned code not in success list
:returns: request response
"""
- data = json_data
- if json_data is not None:
- data = json.dumps(json_data)
+ self.set_param('server', server_name, iff=server_name)
+
+ if json_data:
+ json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
- self.set_header('Content-Length', len(data))
+ self.set_header('Content-Length', len(json_data))
+ path = path4url('servers', server_id)
+ return self.put(path, data=json_data, success=success, **kwargs)
- path = path4url('servers', server_id, command)
- return self.put(path, data=data, success=success, **kwargs)
+ def servers_delete(self, server_id, success=204, **kwargs):
+ """DEL ETE base_url/servers/<server_id>
- def flavors_get(self, flavor_id='', command='', success=200, **kwargs):
- """GET base_url[/flavor_id][/command]
+ :param json_data: a json-formated dict that will be send as data
- :param flavor_id: integer (str or int)
+ :param success: success code (iterable of) codes
- :param command: flavor service command
+ :raises ClientError: if returned code not in success list
- :param success: success code or list or tupple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ :returns: request response
+ """
+ path = path4url('servers', server_id)
+ return self.delete(path, success=success, **kwargs)
+
+ def servers_metadata_get(self, server_id, key=None, success=200, **kwargs):
+ """GET base_url/servers/<server_id>/metadata[/key]
:returns: request response
"""
- path = path4url('flavors', flavor_id, command)
+ path = path4url('servers', server_id, 'metadata', key or '')
return self.get(path, success=success, **kwargs)
- def images_get(self, image_id='', command='', success=200, **kwargs):
- """GET base_url[/image_id][/command]
+ def servers_metadata_post(
+ self, server_id, json_data=None, success=202, **kwargs):
+ """POST base_url/servers/<server_id>/metadata
+
+ :returns: request response
+ """
+ if json_data:
+ json_data = json.dumps(json_data)
+ self.set_header('Content-Type', 'application/json')
+ self.set_header('Content-Length', len(json_data))
+ path = path4url('servers', server_id, 'metadata')
+ return self.post(path, data=json_data, success=success, **kwargs)
- :param image_id: string
+ def servers_metadata_put(
+ self, server_id, key=None, json_data=None, success=204, **kwargs):
+ """PUT base_url/servers/<server_id>/metadata[/key]
- :param command: image server command
+ :returns: request response
+ """
+ if json_data:
+ json_data = json.dumps(json_data)
+ self.set_header('Content-Type', 'application/json')
+ self.set_header('Content-Length', len(json_data))
+ path = path4url('servers', server_id, 'metadata', key or '')
+ return self.put(path, data=json_data, success=success, **kwargs)
- :param success: success code or list or tupple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ def servers_metadata_delete(self, server_id, key, success=204, **kwargs):
+ """DEL ETE base_url/servers/<server_id>/metadata[/key]
:returns: request response
"""
- path = path4url('images', image_id, command)
- return self.get(path, success=success, **kwargs)
+ path = path4url('servers', server_id, 'metadata', key)
+ return self.delete(path, success=success, **kwargs)
- def images_delete(self, image_id='', command='', success=204, **kwargs):
- """DELETE base_url[/image_id][/command]
+ def servers_action_post(
+ self, server_id, json_data=None, success=202, **kwargs):
+ """POST base_url/servers/<server_id>/action
- :param image_id: string
+ :returns: request response
+ """
+ if json_data:
+ json_data = json.dumps(json_data)
+ self.set_header('Content-Type', 'application/json')
+ self.set_header('Content-Length', len(json_data))
+ path = path4url('servers', server_id, 'action')
+ return self.post(path, data=json_data, success=success, **kwargs)
- :param command: image server command
+ def servers_ips_get(
+ self, server_id,
+ network_id=None, changes_since=None, success=(304, 200),
+ **kwargs):
+ """GET base_url/servers/<server_id>/ips[/network_id]
- :param success: success code or list or tuple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ :param changes_since: time/date stamp in UNIX/epoch time. Checks for
+ changes since a previous request.
:returns: request response
"""
- path = path4url('images', image_id, command)
- return self.delete(path, success=success, **kwargs)
+ self.set_param('changes-since', changes_since, iff=changes_since)
+ path = path4url('servers', server_id, 'ips', network_id or '')
+ return self.get(path, success=success, **kwargs)
- def images_post(
+ def images_get(
self,
image_id='',
- command='',
- json_data=None,
- success=201,
+ detail=False,
+ changes_since=None,
+ server_name=None,
+ name=None,
+ status=None,
+ marker=None,
+ limit=None,
+ type=None,
+ success=200,
**kwargs):
- """POST base_url/images[/image_id]/[command] request
+ """GET base_url[/image_id][/command]
- :param image_id: string
+ :param image_id: (str) ignored if detail
- :param command: image server command
+ :param detail: (bool)
- :param json_data: (dict) will be send as data
+ --- Parameters ---
+ :param changes_since: when the image last changed status
- :param success: success code or list or tuple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ :param server_name: Name of the server in URL format
+
+ :param name: Name of the image
+
+ :param status: Status of the image (e.g. filter on "ACTIVE")
+
+ :param marker: UUID of the image at which you want to set a marker
+
+ :param limit: Integer value for the limit of values to return
+
+ :param type: Type of image (e.g. BASE, SERVER, or ALL)
+
+ :returns: request response
+ """
+ if not image_id:
+ self.set_param('changes-since', changes_since, iff=changes_since)
+ self.set_param('name', name, iff=name)
+ self.set_param('status', status, iff=status)
+ self.set_param('marker', marker, iff=marker)
+ self.set_param('limit', limit, iff=limit)
+ self.set_param('type', type, iff=type)
+ self.set_param('server', server_name, iff=server_name)
+
+ path = path4url('images', 'detail' if detail else (image_id or ''))
+ return self.get(path, success=success, **kwargs)
+
+ def images_delete(self, image_id='', success=204, **kwargs):
+ """DEL ETE base_url/images/<image_id>
+
+ :returns: request response
+ """
+ path = path4url('images', image_id)
+ return self.delete(path, success=success, **kwargs)
+
+ def images_metadata_get(self, image_id, key=None, success=200, **kwargs):
+ """GET base_url/<image_id>/metadata[/key]
+
+ :returns: request response
+ """
+ path = path4url('images', image_id, 'metadata', key or '')
+ return self.get(path, success=success, **kwargs)
+
+ def images_metadata_post(
+ self, image_id, json_data=None, success=201, **kwargs):
+ """POST base_url/images/<image_id>/metadata
:returns: request response
"""
- data = json_data
if json_data is not None:
- data = json.dumps(json_data)
+ json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
- self.set_header('Content-Length', len(data))
+ self.set_header('Content-Length', len(json_data))
- path = path4url('images', image_id, command)
- return self.post(path, data=data, success=success, **kwargs)
+ path = path4url('images', image_id, 'metadata')
+ return self.post(path, data=json_data, success=success, **kwargs)
+
+ def images_metadata_put(
+ self, image_id, key=None, json_data=None, success=201, **kwargs):
+ """PUT base_url/images/<image_id>/metadata
- def images_put(
+ :returns: request response
+ """
+ if json_data is not None:
+ json_data = json.dumps(json_data)
+ self.set_header('Content-Type', 'application/json')
+ self.set_header('Content-Length', len(json_data))
+
+ path = path4url('images', image_id, 'metadata')
+ return self.put(path, data=json_data, success=success, **kwargs)
+
+ def images_metadata_delete(self, image_id, key, success=204, **kwargs):
+ """DEL ETE base_url/images/<image_id>/metadata/key
+
+ :returns: request response
+ """
+ path = path4url('images', image_id, 'metadata', key)
+ return self.delete(path, success=success, **kwargs)
+
+ def flavors_get(
self,
- image_id='',
- command='',
- json_data=None,
- success=201,
+ flavor_id='',
+ detail=False,
+ changes_since=None,
+ minDisk=None,
+ minRam=None,
+ marker=None,
+ limit=None,
+ success=200,
**kwargs):
- """PUT base_url/images[/image_id]/[command] request
+ """GET base_url[/flavor_id][/command]
- :param image_id: string
+ :param flavor_id: ignored if detail
- :param command: image server command
+ :param detail: (bool)
- :param json_data: (dict) will be send as data
+ --- Parameters ---
- :param success: success code or list or tuple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
+ :param changes_since: when the flavor last changed
+
+ :param minDisk: minimum disk space in GB filter
+
+ :param minRam: minimum RAM filter
+
+ :param marker: UUID of the flavor at which to set a marker
+
+ :param limit: limit the number of returned values
:returns: request response
"""
- data = json_data
- if json_data is not None:
- data = json.dumps(json_data)
- self.set_header('Content-Type', 'application/json')
- self.set_header('Content-Length', len(data))
-
- path = path4url('images', image_id, command)
- return self.put(path, data=data, success=success, **kwargs)
+ if not flavor_id:
+ self.set_param('changes-since', changes_since, iff=changes_since)
+ self.set_param('minDisk', minDisk, iff=minDisk)
+ self.set_param('minRam', minRam, iff=minRam)
+ self.set_param('marker', marker, iff=marker)
+ self.set_param('limit', limit, iff=limit)
+
+ path = path4url('flavors', 'detail' if detail else (flavor_id or ''))
+ return self.get(path, success=success, **kwargs)
- def floating_ip_pools_get(self, tenant_id, success=200, **kwargs):
- path = path4url(tenant_id, 'os-floating-ip-pools')
+ def floating_ip_pools_get(self, success=200, **kwargs):
+ path = path4url('os-floating-ip-pools')
return self.get(path, success=success, **kwargs)
- def floating_ips_get(self, tenant_id, ip='', success=200, **kwargs):
- path = path4url(tenant_id, 'os-floating-ips', ip or '')
+ def floating_ips_get(self, ip='', success=200, **kwargs):
+ path = path4url('os-floating-ips', ip or '')
return self.get(path, success=success, **kwargs)
- def floating_ips_post(
- self, tenant_id, json_data, ip='', success=201, **kwargs):
- path = path4url(tenant_id, 'os-floating-ips', ip or '')
+ def floating_ips_post(self, json_data, ip='', success=201, **kwargs):
+ path = path4url('os-floating-ips', ip or '')
if json_data is not None:
json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
self.set_header('Content-Length', len(json_data))
return self.post(path, data=json_data, success=success, **kwargs)
- def floating_ips_delete(self, tenant_id, ip='', success=204, **kwargs):
- path = path4url(tenant_id, 'os-floating-ips', ip or '')
+ def floating_ips_delete(self, ip='', success=204, **kwargs):
+ path = path4url('os-floating-ips', ip or '')
return self.delete(path, success=success, **kwargs)
from unittest import TestCase
from itertools import product
from json import dumps
+from sys import stdout
+import json
from kamaki.clients import ClientError, compute
status_code = 200
+def print_iterations(old, new):
+ if new:
+ if new % 1000:
+ return old
+ stdout.write('\b' * len('%s' % old))
+ stdout.write('%s' % new)
+ else:
+ stdout.write('# of loops: ')
+ stdout.flush()
+ return new
+
+
class ComputeRestClient(TestCase):
"""Set up a ComputesRest thorough test"""
def tearDown(self):
FR.json = vm_recv
+ @patch('%s.set_param' % rest_pkg)
@patch('%s.get' % rest_pkg, return_value=FR())
- def _test_get(self, service, get):
- for args in product(
+ def _test_get(self, service, params, get, set_param):
+ method = getattr(self.client, '%s_get' % service)
+ param_args = [({}, {k: k}, {k: v[1]}) for k, v in params.items()]
+ num_of_its = 0
+ for i, args in enumerate(product(
('', '%s_id' % service),
- ('', 'cmd'),
+ (None, False, True),
(200, 204),
- ({}, {'k': 'v'})):
- (srv_id, command, success, kwargs) = args
- method = getattr(self.client, '%s_get' % service)
- method(*args[:3], **kwargs)
- srv_str = '/%s' % srv_id if srv_id else ''
- cmd_str = '/%s' % command if command else ''
- self.assertEqual(get.mock_calls[-1], call(
- '/%s%s%s' % (service, srv_str, cmd_str),
- success=success,
- **kwargs))
-
- def test_servers_get(self):
- self._test_get('servers')
-
- def test_flavors_get(self):
- self._test_get('flavors')
-
- def test_images_get(self):
- self._test_get('images')
-
- @patch('%s.delete' % rest_pkg, return_value=FR())
- def _test_delete(self, service, delete):
- for args in product(
- ('', '%s_id' % service),
- ('', 'cmd'),
- (204, 208),
- ({}, {'k': 'v'})):
- (srv_id, command, success, kwargs) = args
- method = getattr(self.client, '%s_delete' % service)
- method(*args[:3], **kwargs)
- vm_str = '/%s' % srv_id if srv_id else ''
- cmd_str = '/%s' % command if command else ''
- self.assertEqual(delete.mock_calls[-1], call(
- '/%s%s%s' % (service, vm_str, cmd_str),
- success=success,
- **kwargs))
-
- def test_servers_delete(self):
- self._test_delete('servers')
-
- def test_images_delete(self):
- self._test_delete('images')
+ ({}, {'k': 'v'}),
+ *param_args)):
+ (srv_id, detail, success, kwargs) = args[:4]
+ kwargs['success'] = success
+ srv_kwargs = dict()
+ for param in args[4:]:
+ srv_kwargs.update(param)
+ srv_kwargs.update(kwargs)
+ method(*args[:2], **srv_kwargs)
+ srv_str = '/detail' if detail else (
+ '/%s' % srv_id) if srv_id else ''
+ self.assertEqual(
+ get.mock_calls[-1],
+ call('/%s%s' % (service, srv_str), **kwargs))
+ param_calls = []
+ for k, v in params.items():
+ real_v = srv_kwargs.get(k, v[1]) if not srv_id else v[1]
+ param_calls.append(call(v[0], real_v, iff=real_v))
+ actual = set_param.mock_calls[- len(param_calls):]
+ self.assertEqual(sorted(actual), sorted(param_calls))
+
+ num_of_its = print_iterations(num_of_its, i)
+ print ('\b' * len('%s' % num_of_its)) + ('%s' % i)
+
+ @patch('%s.set_param' % rest_pkg)
+ @patch('%s.get' % rest_pkg, return_value=FR())
+ def _test_srv_cmd_get(self, srv, cmd, params, get, set_param):
+ method = getattr(self.client, '%s_%s_get' % (srv, cmd))
+ param_args = [({}, {k: k}, {k: v[1]}) for k, v in params.items()]
+ num_of_its = 0
+ for i, args in enumerate(product(
+ ('some_server_id', 'other_server_id'),
+ (None, 'xtra_id'),
+ ((304, 200), (1000)),
+ ({}, {'k': 'v'}),
+ *param_args)):
+ srv_id, xtra_id, success, kwargs = args[:4]
+ kwargs = dict(kwargs)
+ kwargs['success'] = success
+ srv_kwargs = dict()
+ for param in args[4:]:
+ srv_kwargs.update(param)
+ srv_kwargs.update(kwargs)
+ method(*args[:2], **srv_kwargs)
+ srv_str = '/%s/%s/%s' % (srv, srv_id, cmd)
+ srv_str += ('/%s' % xtra_id) if xtra_id else ''
+ self.assertEqual(get.mock_calls[-1], call(srv_str, **kwargs))
+ param_calls = []
+ for k, v in params.items():
+ real_v = srv_kwargs.get(k, v[1])
+ param_calls.append(call(v[0], real_v, iff=real_v))
+ actual = set_param.mock_calls[- len(param_calls):]
+ self.assertEqual(sorted(actual), sorted(param_calls))
+
+ num_of_its = print_iterations(num_of_its, i)
+ print ('\b' * len('%s' % num_of_its)) + ('%s' % i)
@patch('%s.set_header' % rest_pkg)
+ @patch('%s.set_param' % rest_pkg)
@patch('%s.post' % rest_pkg, return_value=FR())
- def _test_post(self, service, post, SH):
- for args in product(
- ('', '%s_id' % service),
- ('', 'cmd'),
- (None, [dict(json="data"), dict(data="json")]),
- (202, 204),
- ({}, {'k': 'v'})):
- (srv_id, command, json_data, success, kwargs) = args
- method = getattr(self.client, '%s_post' % service)
- method(*args[:4], **kwargs)
- vm_str = '/%s' % srv_id if srv_id else ''
- cmd_str = '/%s' % command if command else ''
+ def _test_post(self, srv, cmd, params, post, set_param, set_header):
+ method = getattr(
+ self.client, '%s_%spost' % (srv, ('%s_' % cmd) if cmd else ''))
+ param_args = [({}, {k: k}, {k: v[1]}) for k, v in params.items()]
+ num_of_its = 0
+ for i, args in enumerate(product(
+ ('%s_id' % srv, 'some_value'),
+ (
+ None,
+ {'some': {'data': 'in json'}},
+ ['k1', {'k2': 'v2', 'k3': 'v3'}, 'k4']),
+ (202, 1453),
+ ({}, {'k': 'v'}),
+ *param_args)):
+ srv_id, json_data, success, kwargs = args[:4]
+ kwargs = dict(kwargs)
+ cmd_args = (srv_id, ) if cmd else ()
+ kwargs['success'] = success
+ srv_kwargs = dict()
+ for param in args[4:]:
+ srv_kwargs.update(param)
+ srv_kwargs.update(kwargs)
+ srv_kwargs['json_data'] = json_data
+ method(*cmd_args, **srv_kwargs)
+ srv_str = '/%s%s' % (
+ srv, (('/%s/%s' % (srv_id, cmd)) if cmd else ''))
+ kwargs['data'] = json.dumps(json_data) if json_data else None
+ self.assertEqual(post.mock_calls[-1], call(srv_str, **kwargs))
+
+ param_calls = []
+ for k, v in params.items():
+ real_v = srv_kwargs.get(k, v[1])
+ param_calls.append(call(v[0], real_v, iff=real_v))
+ actual = set_param.mock_calls[- len(param_calls):]
+ self.assertEqual(sorted(actual), sorted(param_calls))
+
if json_data:
- json_data = dumps(json_data)
- self.assertEqual(SH.mock_calls[-2:], [
+ self.assertEqual(set_header.mock_calls[-2:], [
call('Content-Type', 'application/json'),
- call('Content-Length', len(json_data))])
- self.assertEqual(post.mock_calls[-1], call(
- '/%s%s%s' % (service, vm_str, cmd_str),
- data=json_data, success=success,
- **kwargs))
+ call('Content-Length', len(kwargs['data']))])
- def test_servers_post(self):
- self._test_post('servers')
-
- def test_images_post(self):
- self._test_post('images')
+ num_of_its = print_iterations(num_of_its, i)
+ print ('\b' * len('%s' % num_of_its)) + ('%s' % i)
@patch('%s.set_header' % rest_pkg)
+ @patch('%s.set_param' % rest_pkg)
@patch('%s.put' % rest_pkg, return_value=FR())
- def _test_put(self, service, put, SH):
- for args in product(
- ('', '%s_id' % service),
- ('', 'cmd'),
+ def _test_put(self, srv, cmd, params, put, set_param, set_headers):
+ method = getattr(self.client, '%s_%sput' % (
+ srv, ('%s_' % cmd) if cmd else ''))
+ param_args = [({}, {k: k}, {k: v[1]}) for k, v in params.items()]
+ num_of_its = 0
+ for i, args in enumerate(product(
+ ('some_value', '%s_id' % srv),
(None, [dict(json="data"), dict(data="json")]),
(204, 504),
- ({}, {'k': 'v'})):
- (server_id, command, json_data, success, kwargs) = args
- method = getattr(self.client, '%s_put' % service)
- method(*args[:4], **kwargs)
- vm_str = '/%s' % server_id if server_id else ''
- cmd_str = '/%s' % command if command else ''
+ ({}, {'k': 'v'}),
+ *param_args)):
+ srv_id, json_data, success, kwargs = args[:4]
+ kwargs = dict(kwargs)
+ kwargs['success'] = success
+ srv_kwargs = dict()
+ for param in args[4:]:
+ srv_kwargs.update(param)
+ srv_kwargs.update(kwargs)
+ srv_kwargs['json_data'] = json_data
+ method(srv_id, **srv_kwargs)
+ srv_str = '/%s/%s%s' % (srv, srv_id, ('/%s' % cmd if cmd else ''))
+
if json_data:
json_data = dumps(json_data)
- self.assertEqual(SH.mock_calls[-2:], [
+ self.assertEqual(set_headers.mock_calls[-2:], [
call('Content-Type', 'application/json'),
call('Content-Length', len(json_data))])
- self.assertEqual(put.mock_calls[-1], call(
- '/%s%s%s' % (service, vm_str, cmd_str),
- data=json_data, success=success,
- **kwargs))
+ self.assertEqual(
+ put.mock_calls[-1],
+ call(srv_str, data=json_data, **kwargs))
+
+ param_calls = []
+ for k, v in params.items():
+ real_v = srv_kwargs.get(k, v[1])
+ param_calls.append(call(v[0], real_v, iff=real_v))
+ actual = set_param.mock_calls[- len(param_calls):]
+ self.assertEqual(sorted(actual), sorted(param_calls))
+
+ num_of_its = print_iterations(num_of_its, i)
+ print ('\b' * len('%s' % num_of_its)) + ('%s' % i)
+
+ @patch('%s.delete' % rest_pkg, return_value=FR())
+ def _test_delete(self, srv, cmd, delete):
+ method = getattr(
+ self.client, '%s_%sdelete' % (srv, ('%s_' % cmd) if cmd else ''))
+ cmd_params = ('some_cmd_value', 'some_other_value') if cmd else ()
+ num_of_its = 0
+ for i, args in enumerate(product(
+ ('%s_id' % srv, 'some_value'),
+ (204, 208),
+ ({}, {'k': 'v'}),
+ *cmd_params)):
+ (srv_id, success, kwargs) = args[:3]
+ kwargs = dict(kwargs)
+ kwargs['success'] = success
+ cmd_value = args[-1] if cmd else ''
+ method_args = (srv_id, cmd_value) if cmd else (srv_id, )
+ method(*method_args, **kwargs)
+ srv_str = '/%s/%s' % (srv, srv_id)
+ cmd_str = ('/%s/%s' % (cmd, cmd_value)) if cmd else ''
+ self.assertEqual(
+ delete.mock_calls[-1],
+ call('%s%s' % (srv_str, cmd_str), **kwargs))
+ num_of_its = print_iterations(num_of_its, i)
+ print ('\b' * len('%s' % num_of_its)) + ('%s' % i)
+
+ @patch('%s.get' % rest_pkg, return_value=FR())
+ def test_limits_get(self, get):
+ self.client.limits_get(success='some_val')
+ get.assert_called_once_with('/limits', success='some_val')
+
+ def test_servers_get(self):
+ params = dict(
+ changes_since=('changes-since', None),
+ image=('image', None),
+ flavor=('flavor', None),
+ name=('name', None),
+ marker=('marker', None),
+ limit=('limit', None),
+ status=('status', None),
+ host=('host', None))
+ self._test_get('servers', params)
+
+ def test_servers_post(self):
+ params = dict(
+ security_group=('security_group', None),
+ user_data=('user_data', None),
+ availability_zone=('availability_zone', None))
+ self._test_post('servers', None, params)
def test_servers_put(self):
- self._test_put('servers')
+ self._test_put('servers', None, dict(server_name=('server', None)))
+
+ def test_servers_delete(self):
+ self._test_delete('servers', None)
+
+ def test_servers_metadata_get(self):
+ self._test_srv_cmd_get('servers', 'metadata', {})
+
+ def test_servers_metadata_post(self):
+ self._test_post('servers', 'metadata', {})
+
+ def test_servers_metadata_put(self):
+ self._test_put('servers', 'metadata', {})
+
+ def test_images_metadata_delete(self):
+ self._test_delete('images', 'metadata')
- def test_images_put(self):
- self._test_put('images')
+ def test_servers_action_post(self):
+ self._test_post('servers', 'action', {})
+
+ def test_servers_ips_get(self):
+ params = dict(changes_since=('changes-since', None))
+ self._test_srv_cmd_get('servers', 'ips', params)
+
+ def test_images_get(self):
+ param = dict(
+ changes_since=('changes-since', None),
+ server_name=('server', None),
+ name=('name', None),
+ status=('status', None),
+ marker=('marker', None),
+ limit=('limit', None),
+ type=('type', None))
+ self._test_get('images', param)
+
+ def test_images_delete(self):
+ self._test_delete('images', None)
+
+ def test_images_metadata_get(self):
+ self._test_srv_cmd_get('images', 'metadata', {})
+
+ def test_images_metadata_post(self):
+ self._test_post('images', 'metadata', {})
+
+ def test_images_metadata_put(self):
+ self._test_put('images', 'metadata', {})
+
+ def test_servers_metadata_delete(self):
+ self._test_delete('servers', 'metadata')
+
+ def test_flavors_get(self):
+ params = dict(
+ changes_since=('changes-since', None),
+ minDisk=('minDisk', None),
+ minRam=('minRam', None),
+ marker=('marker', None),
+ limit=('limit', None))
+ self._test_get('flavors', params)
@patch('%s.get' % rest_pkg, return_value=FR())
def test_floating_ip_pools_get(self, get):
for args in product(
- ('tenant1', 'tenant2'),
(200, 204),
({}, {'k': 'v'})):
- tenant_id, success, kwargs = args
- r = self.client.floating_ip_pools_get(tenant_id, success, **kwargs)
+ success, kwargs = args
+ r = self.client.floating_ip_pools_get(success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(get.mock_calls[-1], call(
- '/%s/os-floating-ip-pools' % tenant_id,
- success=success, **kwargs))
+ '/os-floating-ip-pools', success=success, **kwargs))
@patch('%s.get' % rest_pkg, return_value=FR())
def test_floating_ips_get(self, get):
for args in product(
- ('tenant1', 'tenant2'),
('', '192.193.194.195'),
(200, 204),
({}, {'k': 'v'})):
- tenant_id, ip, success, kwargs = args
- r = self.client.floating_ips_get(*args[:3], **kwargs)
+ ip, success, kwargs = args
+ r = self.client.floating_ips_get(*args[:2], **kwargs)
self.assertTrue(isinstance(r, FR))
expected = '' if not ip else '/%s' % ip
self.assertEqual(get.mock_calls[-1], call(
- '/%s/os-floating-ips%s' % (tenant_id, expected),
- success=success, **kwargs))
+ '/os-floating-ips%s' % expected, success=success, **kwargs))
@patch('%s.set_header' % rest_pkg)
@patch('%s.post' % rest_pkg, return_value=FR())
def test_floating_ips_post(self, post, SH):
for args in product(
- ('tenant1', 'tenant2'),
(None, [dict(json="data"), dict(data="json")]),
('', '192.193.194.195'),
(202, 204),
({}, {'k': 'v'})):
- (tenant_id, json_data, ip, success, kwargs) = args
- self.client.floating_ips_post(*args[:4], **kwargs)
+ json_data, ip, success, kwargs = args
+ self.client.floating_ips_post(*args[:3], **kwargs)
if json_data:
json_data = dumps(json_data)
self.assertEqual(SH.mock_calls[-2:], [
call('Content-Length', len(json_data))])
expected = '' if not ip else '/%s' % ip
self.assertEqual(post.mock_calls[-1], call(
- '/%s/os-floating-ips%s' % (tenant_id, expected),
+ '/os-floating-ips%s' % expected,
data=json_data, success=success,
**kwargs))
@patch('%s.delete' % rest_pkg, return_value=FR())
def test_floating_ips_delete(self, delete):
for args in product(
- ('tenant1', 'tenant2'),
('', '192.193.194.195'),
(204,),
({}, {'k': 'v'})):
- tenant_id, ip, success, kwargs = args
- r = self.client.floating_ips_delete(*args[:3], **kwargs)
+ ip, success, kwargs = args
+ r = self.client.floating_ips_delete(*args[:2], **kwargs)
self.assertTrue(isinstance(r, FR))
expected = '' if not ip else '/%s' % ip
self.assertEqual(delete.mock_calls[-1], call(
- '/%s/os-floating-ips%s' % (tenant_id, expected),
- success=success, **kwargs))
+ '/os-floating-ips%s' % expected, success=success, **kwargs))
class ComputeClient(TestCase):
FR.status_code = 200
FR.json = vm_recv
- @patch(
- '%s.get_image_details' % compute_pkg,
- return_value=img_recv['image'])
- def test_create_server(self, GID):
+ def test_create_server(self):
with patch.object(
compute.ComputeClient, 'servers_post',
side_effect=ClientError(
self.client.create_server,
vm_name, fid, img_ref)
- with patch.object(
- compute.ComputeClient, 'servers_post',
- return_value=FR()) as post:
- r = self.client.create_server(vm_name, fid, img_ref)
- self.assertEqual(r, FR.json['server'])
- self.assertEqual(GID.mock_calls[-1], call(img_ref))
- self.assertEqual(post.mock_calls[-1], call(json_data=vm_send))
- prsn = 'Personality string (does not work with real servers)'
- self.client.create_server(vm_name, fid, img_ref, prsn)
- expected = dict(server=dict(vm_send['server']))
- expected['server']['personality'] = prsn
- self.assertEqual(post.mock_calls[-1], call(json_data=expected))
+ for params in product(
+ ('security_group', None),
+ ('user_data', None),
+ ('availability_zone', None),
+ (None, {'os': 'debian', 'users': 'root'})):
+ kwargs = dict()
+ for i, k in enumerate((
+ 'security_group', 'user_data', 'availability_zone')):
+ if params[i]:
+ kwargs[k] = params[i]
+ with patch.object(
+ compute.ComputeClient, 'servers_post',
+ return_value=FR()) as post:
+ r = self.client.create_server(vm_name, fid, img_ref, **kwargs)
+ self.assertEqual(r, FR.json['server'])
+ exp_json = dict(server=dict(
+ flavorRef=fid, name=vm_name, imageRef=img_ref))
+ for k in set([
+ 'security_group',
+ 'user_data',
+ 'availability_zone']).difference(kwargs):
+ kwargs[k] = None
+ self.assertEqual(
+ post.mock_calls[-1], call(json_data=exp_json, **kwargs))
+ prsn = 'Personality string (does not work with real servers)'
+ self.client.create_server(
+ vm_name, fid, img_ref, personality=prsn, **kwargs)
+ exp_json['server']['personality'] = prsn
+ self.assertEqual(
+ post.mock_calls[-1], call(json_data=exp_json, **kwargs))
+ kwargs.pop('personality', None)
+ exp_json['server'].pop('personality', None)
+ mtdt = 'Metadata dict here'
+ self.client.create_server(
+ vm_name, fid, img_ref, metadata=mtdt, **kwargs)
+ exp_json['server']['metadata'] = mtdt
+ self.assertEqual(
+ post.mock_calls[-1], call(json_data=exp_json, **kwargs))
@patch('%s.servers_get' % compute_pkg, return_value=FR())
def test_list_servers(self, SG):
FR.json = vm_list
- for detail in (False, True):
- r = self.client.list_servers(detail)
- self.assertEqual(SG.mock_calls[-1], call(
- command='detail' if detail else ''))
+ for args in product(
+ (False, True),
+ ({}, dict(status='status')),
+ ({}, dict(name='name')),
+ ({}, dict(image='image')),
+ ({}, dict(flavor='flavor')),
+ ({}, dict(host='host')),
+ ({}, dict(limit='limit')),
+ ({}, dict(marker='marker')),
+ ({}, dict(changes_since='changes_since'))):
+ detail = args[0]
+ kwargs = dict()
+ for param in args[1:]:
+ kwargs.update(param)
+ r = self.client.list_servers(detail, **kwargs)
+ for k in set([
+ 'status', 'name',
+ 'image', 'flavor',
+ 'host', 'limit',
+ 'marker', 'changes_since']).difference(kwargs):
+ kwargs[k] = None
+ self.assertEqual(SG.mock_calls[-1], call(detail=detail, **kwargs))
for i, vm in enumerate(vm_list['servers']):
self.assert_dicts_are_equal(r[i], vm)
self.assertEqual(i + 1, len(r))
@patch('%s.servers_get' % compute_pkg, return_value=FR())
def test_get_server_details(self, SG):
vm_id = vm_recv['server']['id']
- r = self.client.get_server_details(vm_id)
- SG.assert_called_once_with(vm_id)
- self.assert_dicts_are_equal(r, vm_recv['server'])
+ for args in product(
+ ({}, dict(status='status')),
+ ({}, dict(name='name')),
+ ({}, dict(image='image')),
+ ({}, dict(flavor='flavor')),
+ ({}, dict(host='host')),
+ ({}, dict(limit='limit')),
+ ({}, dict(marker='marker')),
+ ({}, dict(changes_since='changes_since'))):
+ kwargs = dict()
+ for param in args:
+ kwargs.update(param)
+ r = self.client.get_server_details(vm_id, **kwargs)
+ for k in set([
+ 'status', 'name',
+ 'image', 'flavor',
+ 'host', 'limit',
+ 'marker', 'changes_since']).difference(kwargs):
+ kwargs[k] = None
+ self.assertEqual(SG.mock_calls[-1], call(vm_id, **kwargs))
+ self.assert_dicts_are_equal(r, vm_recv['server'])
@patch('%s.servers_put' % compute_pkg, return_value=FR())
def test_update_server_name(self, SP):
SP.assert_called_once_with(vm_id, json_data=dict(
server=dict(name=new_name)))
- @patch('%s.servers_post' % compute_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % compute_pkg, return_value=FR())
def test_reboot_server(self, SP):
vm_id = vm_recv['server']['id']
for hard in (None, True):
self.client.reboot_server(vm_id, hard=hard)
- self.assertEqual(SP.mock_calls[-1], call(
- vm_id, 'action',
- json_data=dict(reboot=dict(type='HARD' if hard else 'SOFT'))))
+ self.assertEqual(SP.mock_calls[-1], call(vm_id, json_data=dict(
+ reboot=dict(type='HARD' if hard else 'SOFT'))))
- @patch('%s.servers_post' % compute_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % compute_pkg, return_value=FR())
def test_resize_server(self, SP):
vm_id, flavor = vm_recv['server']['id'], flavor_list['flavors'][1]
self.client.resize_server(vm_id, flavor['id'])
exp = dict(resize=dict(flavorRef=flavor['id']))
- SP.assert_called_once_with(vm_id, 'action', json_data=exp)
+ SP.assert_called_once_with(vm_id, json_data=exp)
- @patch('%s.servers_put' % compute_pkg, return_value=FR())
+ @patch('%s.servers_metadata_put' % compute_pkg, return_value=FR())
def test_create_server_metadata(self, SP):
vm_id = vm_recv['server']['id']
metadata = dict(m1='v1', m2='v2', m3='v3')
r = self.client.create_server_metadata(vm_id, k, v)
self.assert_dicts_are_equal(r, vm_recv['server'])
self.assertEqual(SP.mock_calls[-1], call(
- vm_id, 'metadata/%s' % k,
+ vm_id, '%s' % k,
json_data=dict(meta={k: v}), success=201))
- @patch('%s.servers_get' % compute_pkg, return_value=FR())
+ @patch('%s.servers_metadata_get' % compute_pkg, return_value=FR())
def test_get_server_metadata(self, SG):
vm_id = vm_recv['server']['id']
metadata = dict(m1='v1', m2='v2', m3='v3')
FR.json = dict(metadata=metadata)
r = self.client.get_server_metadata(vm_id)
FR.json = dict(meta=metadata)
- SG.assert_called_once_with(vm_id, '/metadata')
+ SG.assert_called_once_with(vm_id, '')
self.assert_dicts_are_equal(r, metadata)
for k, v in metadata.items():
r = self.client.get_server_metadata(vm_id, k)
self.assert_dicts_are_equal(r, {k: v})
self.assertEqual(
- SG.mock_calls[-1], call(vm_id, '/metadata/%s' % k))
+ SG.mock_calls[-1], call(vm_id, '%s' % k))
- @patch('%s.servers_post' % compute_pkg, return_value=FR())
+ @patch('%s.servers_metadata_post' % compute_pkg, return_value=FR())
def test_update_server_metadata(self, SP):
vm_id = vm_recv['server']['id']
metadata = dict(m1='v1', m2='v2', m3='v3')
r = self.client.update_server_metadata(vm_id, **metadata)
self.assert_dicts_are_equal(r, metadata)
SP.assert_called_once_with(
- vm_id, 'metadata',
- json_data=dict(metadata=metadata), success=201)
+ vm_id, json_data=dict(metadata=metadata), success=201)
- @patch('%s.servers_delete' % compute_pkg, return_value=FR())
+ @patch('%s.servers_metadata_delete' % compute_pkg, return_value=FR())
def test_delete_server_metadata(self, SD):
vm_id = vm_recv['server']['id']
key = 'metakey'
self.client.delete_server_metadata(vm_id, key)
- SD.assert_called_once_with(vm_id, 'metadata/' + key)
+ SD.assert_called_once_with(vm_id, key)
@patch('%s.flavors_get' % compute_pkg, return_value=FR())
def test_list_flavors(self, FG):
FR.json = flavor_list
- for cmd in ('', 'detail'):
- r = self.client.list_flavors(detail=(cmd == 'detail'))
- self.assertEqual(FG.mock_calls[-1], call(command=cmd))
+ for detail in ('', 'detail'):
+ r = self.client.list_flavors(detail=bool(detail))
+ self.assertEqual(FG.mock_calls[-1], call(detail=bool(detail)))
self.assertEqual(r, flavor_list['flavors'])
@patch('%s.flavors_get' % compute_pkg, return_value=FR())
@patch('%s.images_get' % compute_pkg, return_value=FR())
def test_list_images(self, IG):
FR.json = img_list
- for cmd in ('', 'detail'):
- r = self.client.list_images(detail=(cmd == 'detail'))
- self.assertEqual(IG.mock_calls[-1], call(command=cmd))
+ for detail in ('', 'detail'):
+ r = self.client.list_images(detail=detail)
+ self.assertEqual(IG.mock_calls[-1], call(detail=bool(detail)))
expected = img_list['images']
for i in range(len(r)):
self.assert_dicts_are_equal(expected[i], r[i])
IG.assert_called_once_with(img_ref)
self.assert_dicts_are_equal(r, img_recv['image'])
- @patch('%s.images_get' % compute_pkg, return_value=FR())
+ @patch('%s.images_metadata_get' % compute_pkg, return_value=FR())
def test_get_image_metadata(self, IG):
for key in ('', '50m3k3y'):
FR.json = dict(meta=img_recv['image']) if (
key) else dict(metadata=img_recv['image'])
r = self.client.get_image_metadata(img_ref, key)
- self.assertEqual(IG.mock_calls[-1], call(
- '%s' % img_ref,
- '/metadata%s' % (('/%s' % key) if key else '')))
+ self.assertEqual(IG.mock_calls[-1], call(img_ref, key or ''))
self.assert_dicts_are_equal(img_recv['image'], r)
@patch('%s.servers_delete' % compute_pkg, return_value=FR())
self.client.delete_image(img_ref)
ID.assert_called_once_with(img_ref)
- @patch('%s.images_put' % compute_pkg, return_value=FR())
+ @patch('%s.images_metadata_put' % compute_pkg, return_value=FR())
def test_create_image_metadata(self, IP):
(key, val) = ('k1', 'v1')
FR.json = dict(meta=img_recv['image'])
r = self.client.create_image_metadata(img_ref, key, val)
IP.assert_called_once_with(
- img_ref, 'metadata/%s' % key,
+ img_ref, '%s' % key,
json_data=dict(meta={key: val}))
self.assert_dicts_are_equal(r, img_recv['image'])
- @patch('%s.images_post' % compute_pkg, return_value=FR())
+ @patch('%s.images_metadata_post' % compute_pkg, return_value=FR())
def test_update_image_metadata(self, IP):
metadata = dict(m1='v1', m2='v2', m3='v3')
FR.json = dict(metadata=metadata)
r = self.client.update_image_metadata(img_ref, **metadata)
IP.assert_called_once_with(
- img_ref, 'metadata',
- json_data=dict(metadata=metadata))
+ img_ref, json_data=dict(metadata=metadata))
self.assert_dicts_are_equal(r, metadata)
- @patch('%s.images_delete' % compute_pkg, return_value=FR())
+ @patch('%s.images_metadata_delete' % compute_pkg, return_value=FR())
def test_delete_image_metadata(self, ID):
key = 'metakey'
self.client.delete_image_metadata(img_ref, key)
- ID.assert_called_once_with(img_ref, '/metadata/%s' % key)
+ ID.assert_called_once_with(img_ref, '%s' % key)
@patch('%s.floating_ip_pools_get' % compute_pkg, return_value=FR())
def test_get_floating_ip_pools(self, get):
class CycladesClient(CycladesRestClient):
"""Synnefo Cyclades Compute API client"""
+ def create_server(
+ self, name, flavor_id, image_id,
+ metadata=None, personality=None):
+ """Submit request to create a new server
+
+ :param name: (str)
+
+ :param flavor_id: integer id denoting a preset hardware configuration
+
+ :param image_id: (str) id denoting the OS image to run on the VM
+
+ :param metadata: (dict) vm metadata updated by os/users image metadata
+
+ :param personality: a list of (file path, file contents) tuples,
+ describing files to be injected into VM upon creation.
+
+ :returns: a dict with the new VMs details
+
+ :raises ClientError: wraps request errors
+ """
+ image = self.get_image_details(image_id)
+ metadata = metadata or dict()
+ for key in ('os', 'users'):
+ try:
+ metadata[key] = image['metadata'][key]
+ except KeyError:
+ pass
+
+ return super(CycladesClient, self).create_server(
+ name, flavor_id, image_id,
+ metadata=metadata, personality=personality)
+
def start_server(self, server_id):
"""Submit a startup request
:returns: (dict) response headers
"""
req = {'start': {}}
- r = self.servers_post(server_id, 'action', json_data=req, success=202)
+ r = self.servers_action_post(server_id, json_data=req, success=202)
return r.headers
def shutdown_server(self, server_id):
:returns: (dict) response headers
"""
req = {'shutdown': {}}
- r = self.servers_post(server_id, 'action', json_data=req, success=202)
+ r = self.servers_action_post(server_id, json_data=req, success=202)
return r.headers
def get_server_console(self, server_id):
:returns: (dict) info to set a VNC connection to VM
"""
req = {'console': {'type': 'vnc'}}
- r = self.servers_post(server_id, 'action', json_data=req, success=200)
+ r = self.servers_action_post(server_id, json_data=req, success=200)
return r.json['console']
def get_firewall_profile(self, server_id):
:returns: (dict) response headers
"""
req = {'firewallProfile': {'profile': profile}}
- r = self.servers_post(server_id, 'action', json_data=req, success=202)
+ r = self.servers_action_post(server_id, json_data=req, success=202)
return r.headers
- def list_servers(self, detail=False, changes_since=None):
- """
- :param detail: (bool) append full server details to each item if true
-
- :param changes_since: (date)
-
- :returns: list of server ids and names
- """
- detail = 'detail' if detail else ''
- r = self.servers_get(command=detail, changes_since=changes_since)
- return r.json['servers']
-
def list_server_nics(self, server_id):
"""
:param server_id: integer (str or int)
:returns: (dict) network interface connections
"""
- r = self.servers_get(server_id, 'ips')
+ r = self.servers_ips_get(server_id)
return r.json['attachments']
- #return r.json['addresses']
def get_server_stats(self, server_id):
"""
:returns: (dict) auto-generated graphs of statistics (urls)
"""
- r = self.servers_get(server_id, 'stats')
+ r = self.servers_stats_get(server_id)
return r.json['stats']
def list_networks(self, detail=False):
"""
server_id = int(server_id)
assert address, 'address is needed for attach_floating_ip'
- r = self.servers_post(
- server_id, 'action',
- json_data=dict(addFloatingIp=dict(address=address)))
+ req = dict(addFloatingIp=dict(address=address))
+ r = self.servers_action_post(server_id, json_data=req)
return r.headers
def detach_floating_ip(self, server_id, address):
"""
server_id = int(server_id)
assert address, 'address is needed for detach_floating_ip'
- r = self.servers_post(
- server_id, 'action',
- json_data=dict(removeFloatingIp=dict(address=address)))
+ req = dict(removeFloatingIp=dict(address=address))
+ r = self.servers_action_post(server_id, json_data=req)
return r.headers
class CycladesRestClient(ComputeClient):
"""Synnefo Cyclades REST API Client"""
- def servers_get(
- self,
- server_id='',
- command='',
- success=200,
- changes_since=None,
- **kwargs):
- """GET base_url/servers[/server_id][/command] request
-
- :param server_id: integer (as int or str)
-
- :param command: 'ips', 'stats', or ''
-
- :param success: success code or list or tupple of accepted success
- codes. if server response code is not in this list, a ClientError
- raises
-
- :param changes_since: (date)
-
- :returns: request response
- """
- path = path4url('servers', server_id, command)
- self.set_param('changes-since', changes_since, changes_since)
+ def servers_stats_get(self, server_id, success=200, **kwargs):
+ """GET base_url/servers/<server_id>/stats"""
+ path = path4url('servers', server_id, 'stats')
return self.get(path, success=success, **kwargs)
def networks_get(
path = path4url('networks', network_id, command)
return self.put(path, data=data, success=success, **kwargs)
-
- def floating_ip_pools_get(self, success=200, **kwargs):
- path = path4url('os-floating-ip-pools')
- return self.get(path, success=success, **kwargs)
-
- def floating_ips_get(self, fip_id='', success=200, **kwargs):
- path = path4url('os-floating-ips', fip_id)
- return self.get(path, success=success, **kwargs)
-
- def floating_ips_post(self, json_data, fip_id='', success=201, **kwargs):
- path = path4url('os-floating-ips', fip_id)
- if json_data is not None:
- json_data = json.dumps(json_data)
- self.set_header('Content-Type', 'application/json')
- self.set_header('Content-Length', len(json_data))
- return self.post(path, data=json_data, success=success, **kwargs)
-
- def floating_ips_delete(self, fip_id, success=200, **kwargs):
- path = path4url('os-floating-ips', fip_id)
- return self.delete(path, success=success, **kwargs)
def tearDown(self):
FR.json = vm_recv
- @patch('%s.set_param' % rest_pkg)
- @patch('%s.get' % rest_pkg, return_value=FR())
- def test_servers_get(self, get, SP):
- for args in product(
- ('', 'vm_id'),
- ('', 'cmd'),
- (200, 204),
- (None, '50m3-d473'),
- ({}, {'k': 'v'})):
- (srv_id, command, success, changes_since, kwargs) = args
- self.client.servers_get(*args[:4], **kwargs)
- srv_str = '/%s' % srv_id if srv_id else ''
- cmd_str = '/%s' % command if command else ''
- self.assertEqual(get.mock_calls[-1], call(
- '/servers%s%s' % (srv_str, cmd_str),
- success=success,
- **kwargs))
- if changes_since:
- self.assertEqual(
- SP.mock_calls[-1],
- call('changes-since', changes_since, changes_since))
-
@patch('%s.get' % rest_pkg, return_value=FR())
def test_networks_get(self, get):
for args in product(
FR.status_code = 200
FR.json = vm_recv
- @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
- def test_list_servers(self, SG):
- FR.json = vm_list
- for detail, since in ((0, 0), (True, 0), (0, 'd473'), (True, 'd473')):
- r = self.client.list_servers(detail=detail, changes_since=since)
- self.assertEqual(SG.mock_calls[-1], call(
- command='detail' if detail else '',
- changes_since=since))
- expected = vm_list['servers']
- for i, vm in enumerate(r):
- self.assert_dicts_are_equal(vm, expected[i])
- self.assertEqual(i + 1, len(expected))
-
- @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
def test_shutdown_server(self, SP):
vm_id = vm_recv['server']['id']
self.client.shutdown_server(vm_id)
SP.assert_called_once_with(
- vm_id, 'action',
- json_data=dict(shutdown=dict()), success=202)
+ vm_id, json_data=dict(shutdown=dict()), success=202)
- @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
def test_start_server(self, SP):
vm_id = vm_recv['server']['id']
self.client.start_server(vm_id)
SP.assert_called_once_with(
- vm_id, 'action',
- json_data=dict(start=dict()), success=202)
+ vm_id, json_data=dict(start=dict()), success=202)
- @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
def test_get_server_console(self, SP):
cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
FR.json = cnsl
vm_id = vm_recv['server']['id']
r = self.client.get_server_console(vm_id)
SP.assert_called_once_with(
- vm_id, 'action',
- json_data=dict(console=dict(type='vnc')), success=200)
+ vm_id, json_data=dict(console=dict(type='vnc')), success=200)
self.assert_dicts_are_equal(r, cnsl['console'])
def test_get_firewall_profile(self):
self.client.get_firewall_profile,
vm_id)
- @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
def test_set_firewall_profile(self, SP):
vm_id = vm_recv['server']['id']
v = firewalls['attachments'][0]['firewallProfile']
self.client.set_firewall_profile(vm_id, v)
- SP.assert_called_once_with(
- vm_id, 'action',
- json_data=dict(firewallProfile=dict(profile=v)), success=202)
-
- @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
- def test_get_server_stats(self, SG):
- vm_id = vm_recv['server']['id']
- stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
- FR.json = dict(stats=stats)
- r = self.client.get_server_stats(vm_id)
- SG.assert_called_once_with(vm_id, 'stats')
- self.assert_dicts_are_equal(stats, r)
+ SP.assert_called_once_with(vm_id, json_data=dict(
+ firewallProfile=dict(profile=v)), success=202)
@patch('%s.networks_post' % cyclades_pkg, return_value=FR())
def test_create_network(self, NP):
json_data=dict(remove=dict(attachment=nic_id)))
self.assertEqual(r, 1)
- @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
+ @patch('%s.servers_ips_get' % cyclades_pkg, return_value=FR())
def test_list_server_nics(self, SG):
vm_id = vm_recv['server']['id']
nics = dict(attachments=[dict(id='nic1'), dict(id='nic2')])
FR.json = nics
r = self.client.list_server_nics(vm_id)
- SG.assert_called_once_with(vm_id, 'ips')
+ SG.assert_called_once_with(vm_id)
expected = nics['attachments']
for i in range(len(r)):
self.assert_dicts_are_equal(r[i], expected[i])
self.assert_dicts_are_equal(r, FR.headers)
self.assertEqual(delete.mock_calls[-1], call(fip))
- @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
def test_attach_floating_ip(self, spost):
vmid, addr = 42, 'anIpAddress'
for err, args in {
r = self.client.attach_floating_ip(vmid, addr)
self.assert_dicts_are_equal(r, FR.headers)
expected = dict(addFloatingIp=dict(address=addr))
- self.assertEqual(
- spost.mock_calls[-1], call(vmid, 'action', json_data=expected))
+ self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
- @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
+ @patch('%s.servers_action_post' % cyclades_pkg, return_value=FR())
def test_detach_floating_ip(self, spost):
vmid, addr = 42, 'anIpAddress'
for err, args in {
r = self.client.detach_floating_ip(vmid, addr)
self.assert_dicts_are_equal(r, FR.headers)
expected = dict(removeFloatingIp=dict(address=addr))
- self.assertEqual(
- spost.mock_calls[-1], call(vmid, 'action', json_data=expected))
+ self.assertEqual(spost.mock_calls[-1], call(vmid, json_data=expected))
if __name__ == '__main__':
projects / kamaki / commitdiff