openstack/python-novaclient
Code Issues Proposed changes

Removed v1.0 support.

Browse Source 
Change-Id: I6850075a2ac0e1558aa94539e73f4fb939dfb318
This commit is contained in:
Rick Harris 2011-12-15 23:10:59 +00:00
parent b3f48f7064
commit 3661be6673
30 changed files with 10 additions and 3295 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
README.rst
View File

@ -161,8 +161,7 @@ You'll find complete documentation on the shell by running
--url AUTH_URL Defaults to env[NOVA_URL] or
https://auth.api.rackspacecloud.com/v1.0
if undefined.
--version VERSION Accepts 1.0 or 1.1, defaults to
env[NOVA_VERSION].
--version VERSION Accepts 1.1, defaults to env[NOVA_VERSION].
--region_name NAME The region name in the Keystone Service Catalog
to use after authentication. Defaults to first
in the list returned.

2
docs/conf.py
View File

@ -47,7 +47,7 @@ copyright = u'Rackspace, based on work by Jacob Kaplan-Moss'
# The short X.Y version.
version = '2.6'
# The full version, including alpha/beta/rc tags.
release = '2.6.7'
release = '2.6.10'
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.

2
docs/shell.rst
View File

@ -41,7 +41,7 @@ For example, in Bash you'd use::
export NOVA_PASSWORD=yadayadayada
export NOVA_PROJECT_ID=myproject
export NOVA_URL=http://...
export NOVA_VERSION=1.0
export NOVA_VERSION=1.1
From there, all shell commands take the form::

2
novaclient/client.py
View File

@ -45,7 +45,7 @@ class HTTPClient(httplib2.Http):
self.password = password
self.projectid = projectid
self.auth_url = auth_url
self.version = 'v1.0'
self.version = 'v1.1'
self.region_name = region_name
self.endpoint_name = endpoint_name

14
novaclient/shell.py
View File

@ -1,5 +1,4 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 OpenStack LLC.
# All Rights Reserved.
#
@ -29,7 +28,6 @@ import sys
from novaclient import base
from novaclient import exceptions as exc
from novaclient import utils
from novaclient.v1_0 import shell as shell_v1_0
from novaclient.v1_1 import shell as shell_v1_1
@ -90,7 +88,7 @@ class OpenStackComputeShell(object):
parser.add_argument('--version',
default=env('NOVA_VERSION'),
help='Accepts 1.0 or 1.1, defaults to env[NOVA_VERSION].')
help='Accepts 1.1, defaults to env[NOVA_VERSION].')
parser.add_argument('--insecure',
default=False,
@ -107,13 +105,11 @@ class OpenStackComputeShell(object):
try:
actions_module = {
'1': shell_v1_0,
'1.0': shell_v1_0,
'1.1': shell_v1_1,
'2': shell_v1_1,
}[version]
except KeyError:
actions_module = shell_v1_0
actions_module = shell_v1_1
self._find_actions(subparsers, actions_module)
self._find_actions(subparsers, self)
@ -238,7 +234,7 @@ class OpenStackComputeShell(object):
else:
password = apikey
if options.version and options.version != '1.0':
if options.version:
if not projectid:
raise exc.CommandError("You must provide an projectid, either "
"via --projectid or via "
@ -267,13 +263,11 @@ class OpenStackComputeShell(object):
def get_api_class(self, version):
try:
return {
"1": shell_v1_0.CLIENT_CLASS,
"1.0": shell_v1_0.CLIENT_CLASS,
"1.1": shell_v1_1.CLIENT_CLASS,
"2": shell_v1_1.CLIENT_CLASS,
}[version]
except KeyError:
return shell_v1_0.CLIENT_CLASS
return shell_v1_1.CLIENT_CLASS
def do_bash_completion(self, args):
"""

1
novaclient/v1_0/__init__.py
View File

@ -1 +0,0 @@
from novaclient.v1_0.client import Client

18
novaclient/v1_0/accounts.py
View File

@ -1,18 +0,0 @@
from novaclient import base
from novaclient.v1_0 import base as local_base
class Account(base.Resource):
pass
class AccountManager(local_base.BootingManagerWithFind):
resource_class = Account
def create_instance_for(self, account_id, name, image, flavor,
ipgroup=None, meta=None, files=None, zone_blob=None,
reservation_id=None):
resource_url = "/accounts/%s/create_instance" % account_id
return self._boot(resource_url, "server", name, image, flavor,
ipgroup=ipgroup, meta=meta, files=files,
zone_blob=zone_blob, reservation_id=reservation_id)

109
novaclient/v1_0/backup_schedules.py
View File

@ -1,109 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
"""
Backup Schedule interface.
"""
from novaclient import base
BACKUP_WEEKLY_DISABLED = 'DISABLED'
BACKUP_WEEKLY_SUNDAY = 'SUNDAY'
BACKUP_WEEKLY_MONDAY = 'MONDAY'
BACKUP_WEEKLY_TUESDAY = 'TUESDAY'
BACKUP_WEEKLY_WEDNESDAY = 'WEDNESDAY'
BACKUP_WEEKLY_THURSDAY = 'THURSDAY'
BACKUP_WEEKLY_FRIDAY = 'FRIDAY'
BACKUP_WEEKLY_SATURDAY = 'SATURDAY'
BACKUP_DAILY_DISABLED = 'DISABLED'
BACKUP_DAILY_H_0000_0200 = 'H_0000_0200'
BACKUP_DAILY_H_0200_0400 = 'H_0200_0400'
BACKUP_DAILY_H_0400_0600 = 'H_0400_0600'
BACKUP_DAILY_H_0600_0800 = 'H_0600_0800'
BACKUP_DAILY_H_0800_1000 = 'H_0800_1000'
BACKUP_DAILY_H_1000_1200 = 'H_1000_1200'
BACKUP_DAILY_H_1200_1400 = 'H_1200_1400'
BACKUP_DAILY_H_1400_1600 = 'H_1400_1600'
BACKUP_DAILY_H_1600_1800 = 'H_1600_1800'
BACKUP_DAILY_H_1800_2000 = 'H_1800_2000'
BACKUP_DAILY_H_2000_2200 = 'H_2000_2200'
BACKUP_DAILY_H_2200_0000 = 'H_2200_0000'
class BackupSchedule(base.Resource):
"""
Represents the daily or weekly backup schedule for some server.
"""
def get(self):
"""
Get this `BackupSchedule` again from the API.
"""
return self.manager.get(server=self.server)
def delete(self):
"""
Delete (i.e. disable and remove) this scheduled backup.
"""
self.manager.delete(server=self.server)
def update(self, enabled=True, weekly=BACKUP_WEEKLY_DISABLED,
daily=BACKUP_DAILY_DISABLED):
"""
Update this backup schedule.
See :meth:`BackupScheduleManager.create` for details.
"""
self.manager.create(self.server, enabled, weekly, daily)
class BackupScheduleManager(base.Manager):
"""
Manage server backup schedules.
"""
resource_class = BackupSchedule
def get(self, server):
"""
Get the current backup schedule for a server.
:arg server: The server (or its ID).
:rtype: :class:`BackupSchedule`
"""
s = base.getid(server)
schedule = self._get('/servers/%s/backup_schedule' % s,
'backupSchedule')
schedule.server = server
return schedule
# Backup schedules use POST for both create and update, so allow both here.
# Unlike the rest of the API, POST here returns no body, so we can't use
# the nice little helper methods.
def create(self, server, enabled=True, weekly=BACKUP_WEEKLY_DISABLED,
daily=BACKUP_DAILY_DISABLED):
"""
Create or update the backup schedule for the given server.
:arg server: The server (or its ID).
:arg enabled: boolean; should this schedule be enabled?
:arg weekly: Run a weekly backup on this day
(one of the `BACKUP_WEEKLY_*` constants)
:arg daily: Run a daily backup at this time
(one of the `BACKUP_DAILY_*` constants)
"""
s = base.getid(server)
body = {'backupSchedule': {
'enabled': enabled, 'weekly': weekly, 'daily': daily
}}
self.api.client.post('/servers/%s/backup_schedule' % s, body=body)
update = create
def delete(self, server):
"""
Remove the scheduled backup for `server`.
:arg server: The server (or its ID).
"""
s = base.getid(server)
self._delete('/servers/%s/backup_schedule' % s)

99
novaclient/v1_0/base.py
View File

@ -1,99 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 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.
"""
Base utilities to build API operation managers and objects on top of.
"""
from novaclient import base
# Python 2.4 compat
try:
all
except NameError:
def all(iterable):
return True not in (not x for x in iterable)
class BootingManagerWithFind(base.ManagerWithFind):
"""Like a `ManagerWithFind`, but has the ability to boot servers."""
def _boot(self, resource_url, response_key, name, image, flavor,
ipgroup=None, meta=None, files=None, zone_blob=None,
reservation_id=None, return_raw=False, min_count=None,
max_count=None):
"""
Create (boot) a new server.
:param name: Something to name the server.
:param image: The :class:`Image` to boot with.
:param flavor: The :class:`Flavor` to boot onto.
:param ipgroup: An initial :class:`IPGroup` for this server.
:param meta: A dict of arbitrary key/value metadata to store for this
server. A maximum of five entries is allowed, and both
keys and values must be 255 characters or less.
:param files: A dict of files to overrwrite on the server upon boot.
Keys are file names (i.e. ``/etc/passwd``) and values
are the file contents (either as a string or as a
file-like object). A maximum of five entries is allowed,
and each file must be 10k or less.
:param zone_blob: a single (encrypted) string which is used internally
by Nova for routing between Zones. Users cannot populate
this field.
:param reservation_id: a UUID for the set of servers being requested.
:param return_raw: If True, don't try to coearse the result into
a Resource object.
"""
body = {"server": {
"name": name,
"imageId": base.getid(image),
"flavorId": base.getid(flavor),
}}
if ipgroup:
body["server"]["sharedIpGroupId"] = base.getid(ipgroup)
if meta:
body["server"]["metadata"] = meta
if reservation_id:
body["server"]["reservation_id"] = reservation_id
if zone_blob:
body["server"]["blob"] = zone_blob
if not min_count:
min_count = 1
if not max_count:
max_count = min_count
body["server"]["min_count"] = min_count
body["server"]["max_count"] = max_count
# Files are a slight bit tricky. They're passed in a "personality"
# list to the POST. Each item is a dict giving a file name and the
# base64-encoded contents of the file. We want to allow passing
# either an open file *or* some contents as files here.
if files:
personality = body['server']['personality'] = []
for filepath, file_or_string in files.items():
if hasattr(file_or_string, 'read'):
data = file_or_string.read()
else:
data = file_or_string
personality.append({
'path': filepath,
'contents': data.encode('base64'),
})
return self._create(resource_url, body, response_key,
return_raw=return_raw)

70
novaclient/v1_0/client.py
View File

@ -1,70 +0,0 @@
from novaclient import client
from novaclient.v1_0 import accounts
from novaclient.v1_0 import backup_schedules
from novaclient.v1_0 import flavors
from novaclient.v1_0 import images
from novaclient.v1_0 import ipgroups
from novaclient.v1_0 import servers
from novaclient.v1_0 import zones
class Client(object):
"""
Top-level object to access the OpenStack Compute API.
Create an instance with your creds::
>>> client = Client(USERNAME, PASSWORD, PROJECT_ID, AUTH_URL)
Then call methods on its managers::
>>> client.servers.list()
...
>>> client.flavors.list()
...
"""
def __init__(self, username, api_key, project_id, auth_url=None,
insecure=False, timeout=None, token=None, region_name=None,
endpoint_name='publicURL', extensions=None):
# FIXME(comstud): Rename the api_key argument above when we
# know it's not being used as keyword argument
password = api_key
self.accounts = accounts.AccountManager(self)
self.backup_schedules = backup_schedules.BackupScheduleManager(self)
self.flavors = flavors.FlavorManager(self)
self.images = images.ImageManager(self)
self.ipgroups = ipgroups.IPGroupManager(self)
self.servers = servers.ServerManager(self)
self.zones = zones.ZoneManager(self)
# Add in any extensions...
if extensions:
for (ext_name, ext_manager_class, ext_module) in extensions:
setattr(self, ext_name, ext_manager_class(self))
_auth_url = auth_url or 'https://auth.api.rackspacecloud.com/v1.0'
self.client = client.HTTPClient(username,
password,
project_id,
_auth_url,
insecure=insecure,
timeout=timeout,
token=token,
region_name=region_name,
endpoint_name=endpoint_name)
def authenticate(self):
"""
Authenticate against the server.
Normally this is called automatically when you first access the API,
but you can call this method to force authentication right now.
Returns on success; raises :exc:`exceptions.Unauthorized` if the
credentials are wrong.
"""
self.client.authenticate()

41
novaclient/v1_0/flavors.py
View File

@ -1,41 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
"""
Flavor interface.
"""
from novaclient import base
class Flavor(base.Resource):
"""
A flavor is an available hardware configuration for a server.
"""
def __repr__(self):
return "<Flavor: %s>" % self.name
class FlavorManager(base.ManagerWithFind):
"""
Manage :class:`Flavor` resources.
"""
resource_class = Flavor
def list(self, detailed=True):
"""
Get a list of all flavors.
:rtype: list of :class:`Flavor`.
"""
detail = ""
if detailed:
detail = "/detail"
return self._list("/flavors%s" % detail, "flavors")
def get(self, flavor):
"""
Get a specific flavor.
:param flavor: The ID of the :class:`Flavor` to get.
:rtype: :class:`Flavor`
"""
return self._get("/flavors/%s" % base.getid(flavor), "flavor")

69
novaclient/v1_0/images.py
View File

@ -1,69 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
"""
Image interface.
"""
from novaclient import base
class Image(base.Resource):
"""
An image is a collection of files used to create or rebuild a server.
"""
def __repr__(self):
return "<Image: %s>" % self.name
def delete(self):
"""
Delete this image.
"""
return self.manager.delete(self)
class ImageManager(base.ManagerWithFind):
"""
Manage :class:`Image` resources.
"""
resource_class = Image
def get(self, image):
"""
Get an image.
:param image: The ID of the image to get.
:rtype: :class:`Image`
"""
return self._get("/images/%s" % base.getid(image), "image")
def list(self, detailed=True):
"""
Get a list of all images.
:rtype: list of :class:`Image`
"""
detail = ""
if detailed:
detail = "/detail"
return self._list("/images%s" % detail, "images")
def create(self, server, name):
"""
Create a new image by snapshotting a running :class:`Server`
:param name: An (arbitrary) name for the new image.
:param server: The :class:`Server` (or its ID) to make a snapshot of.
:rtype: :class:`Image`
"""
data = {"image": {"serverId": base.getid(server), "name": name}}
return self._create("/images", data, "image")
def delete(self, image):
"""
Delete an image.
It should go without saying that you can't delete an image
that you didn't create.
:param image: The :class:`Image` (or its ID) to delete.
"""
self._delete("/images/%s" % base.getid(image))

64
novaclient/v1_0/ipgroups.py
View File

@ -1,64 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
"""
IP Group interface.
"""
from novaclient import base
class IPGroup(base.Resource):
def __repr__(self):
return "<IP Group: %s>" % self.name
def delete(self):
"""
Delete this group.
"""
self.manager.delete(self)
class IPGroupManager(base.ManagerWithFind):
resource_class = IPGroup
def list(self, detailed=True):
"""
Get a list of all groups.
:rtype: list of :class:`IPGroup`
"""
detail = ""
if detailed:
detail = "/detail"
return self._list("/shared_ip_groups%s" % detail, "sharedIpGroups")
def get(self, group):
"""
Get an IP group.
:param group: ID of the image to get.
:rtype: :class:`IPGroup`
"""
return self._get("/shared_ip_groups/%s" % base.getid(group),
"sharedIpGroup")
def create(self, name, server=None):
"""
Create a new :class:`IPGroup`
:param name: An (arbitrary) name for the new image.
:param server: A :class:`Server` (or its ID) to make a member
of this group.
:rtype: :class:`IPGroup`
"""
data = {"sharedIpGroup": {"name": name}}
if server:
data['sharedIpGroup']['server'] = base.getid(server)
return self._create('/shared_ip_groups', data, "sharedIpGroup")
def delete(self, group):
"""
Delete a group.
:param group: The :class:`IPGroup` (or its ID) to delete.
"""
self._delete("/shared_ip_groups/%s" % base.getid(group))

488
novaclient/v1_0/servers.py
View File

@ -1,488 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 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.
"""
Server interface.
"""
import urllib
from novaclient import base
from novaclient.v1_0 import base as local_base
REBOOT_SOFT, REBOOT_HARD = 'SOFT', 'HARD'
class Server(base.Resource):
def __repr__(self):
return "<Server: %s>" % self.name
def delete(self):
"""
Delete (i.e. shut down and delete the image) this server.
"""
self.manager.delete(self)
def update(self, name=None, password=None):
"""
Update the name or the password for this server.
:param name: Update the server's name.
:param password: Update the root password.
"""
self.manager.update(self, name, password)
def share_ip(self, ipgroup, address, configure=True):
"""
Share an IP address from the given IP group onto this server.
:param ipgroup: The :class:`IPGroup` that the given address belongs to.
:param address: The IP address to share.
:param configure: If ``True``, the server will be automatically
configured to use this IP. I don't know why you'd
want this to be ``False``.
"""
self.manager.share_ip(self, ipgroup, address, configure)
def unshare_ip(self, address):
"""
Stop sharing the given address.
:param address: The IP address to stop sharing.
"""
self.manager.unshare_ip(self, address)
def add_fixed_ip(self, network_id):
"""
Add an IP address on a network.
:param network_id: The ID of the network the IP should be on.
"""
self.manager.add_fixed_ip(self, network_id)
def remove_fixed_ip(self, address):
"""
Remove an IP address.
:param address: The IP address to remove.
"""
self.manager.remove_fixed_ip(self, address)
def reboot(self, type=REBOOT_SOFT):
"""
Reboot the server.
:param type: either :data:`REBOOT_SOFT` for a software-level reboot,
or `REBOOT_HARD` for a virtual power cycle hard reboot.
"""
self.manager.reboot(self, type)
def pause(self):
"""
Pause -- Pause the running server.
"""
self.manager.pause(self)
def unpause(self):
"""
Unpause -- Unpause the paused server.
"""
self.manager.unpause(self)
def suspend(self):
"""
Suspend -- Suspend the running server.
"""
self.manager.suspend(self)
def resume(self):
"""
Resume -- Resume the suspended server.
"""
self.manager.resume(self)
def rescue(self):
"""
Rescue -- Rescue the problematic server.
"""
self.manager.rescue(self)
def unrescue(self):
"""
Unrescue -- Unrescue the rescued server.
"""
self.manager.unrescue(self)
def diagnostics(self):
"""Diagnostics -- Retrieve server diagnostics."""
self.manager.diagnostics(self)
def actions(self):
"""Actions -- Retrieve server actions."""
self.manager.actions(self)
def rebuild(self, image):
"""
Rebuild -- shut down and then re-image -- this server.
:param image: the :class:`Image` (or its ID) to re-image with.
"""
self.manager.rebuild(self, image)
def resize(self, flavor):
"""
Resize the server's resources.
:param flavor: the :class:`Flavor` (or its ID) to resize to.
Until a resize event is confirmed with :meth:`confirm_resize`, the old
server will be kept around and you'll be able to roll back to the old
flavor quickly with :meth:`revert_resize`. All resizes are
automatically confirmed after 24 hours.
"""
self.manager.resize(self, flavor)
def backup(self, image_name, backup_type, rotation):
"""
Create a server backup.
:param server: The :class:`Server` (or its ID).
:param image_name: The name to assign the newly create image.
:param backup_type: 'daily' or 'weekly'
:param rotation: number of backups of type 'backup_type' to keep
:returns Newly created :class:`Image` object
"""
return self.manager.backup(self, image_name, backup_type, rotation)
def confirm_resize(self):
"""
Confirm that the resize worked, thus removing the original server.
"""
self.manager.confirm_resize(self)
def revert_resize(self):
"""
Revert a previous resize, switching back to the old server.
"""
self.manager.revert_resize(self)
def migrate(self):
"""
Migrate a server to a new host in the same zone.
"""
self.manager.migrate(self)
@property
def backup_schedule(self):
"""
This server's :class:`BackupSchedule`.
"""
return self.manager.api.backup_schedules.get(self)
@property
def public_ip(self):
"""
Shortcut to get this server's primary public IP address.
"""
if len(self.addresses['public']) == 0:
return ""
return self.addresses['public']
@property
def private_ip(self):
"""
Shortcut to get this server's primary private IP address.
"""
if len(self.addresses['private']) == 0:
return ""
return self.addresses['private']
class ServerManager(local_base.BootingManagerWithFind):
resource_class = Server
def get(self, server):
"""
Get a server.
:param server: ID of the :class:`Server` to get.
:rtype: :class:`Server`
"""
return self._get("/servers/%s" % base.getid(server), "server")
def list(self, detailed=True, search_opts=None):
"""
Get a list of servers.
Optional detailed returns details server info.
Optional reservation_id only returns instances with that
reservation_id.
:rtype: list of :class:`Server`
"""
if search_opts is None:
search_opts = {}
qparams = {}
# only use values in query string if they are set
for opt, val in search_opts.iteritems():
if val:
qparams[opt] = val
query_string = "?%s" % urllib.urlencode(qparams) if qparams else ""
detail = ""
if detailed:
detail = "/detail"
return self._list("/servers%s%s" % (detail, query_string), "servers")
def create(self, name, image, flavor, ipgroup=None, meta=None, files=None,
zone_blob=None, reservation_id=None, min_count=None,
max_count=None):
"""
Create (boot) a new server.
:param name: Something to name the server.
:param image: The :class:`Image` to boot with.
:param flavor: The :class:`Flavor` to boot onto.
:param ipgroup: An initial :class:`IPGroup` for this server.
:param meta: A dict of arbitrary key/value metadata to store for this
server. A maximum of five entries is allowed, and both
keys and values must be 255 characters or less.
:param files: A dict of files to overrwrite on the server upon boot.
Keys are file names (i.e. ``/etc/passwd``) and values
are the file contents (either as a string or as a
file-like object). A maximum of five entries is allowed,
and each file must be 10k or less.
:param zone_blob: a single (encrypted) string which is used internally
by Nova for routing between Zones. Users cannot populate
this field.
:param reservation_id: a UUID for the set of servers being requested.
"""
if not min_count:
min_count = 1
if not max_count:
max_count = min_count
if min_count > max_count:
min_count = max_count
return self._boot("/servers", "server", name, image, flavor,
ipgroup=ipgroup, meta=meta, files=files,
zone_blob=zone_blob, reservation_id=reservation_id,
min_count=min_count, max_count=max_count)
def update(self, server, name=None, password=None):
"""
Update the name or the password for a server.
:param server: The :class:`Server` (or its ID) to update.
:param name: Update the server's name.
:param password: Update the root password.
"""
if name is None and password is None:
return
body = {"server": {}}
if name:
body["server"]["name"] = name
if password:
body["server"]["adminPass"] = password
self._update("/servers/%s" % base.getid(server), body)
def delete(self, server):
"""
Delete (i.e. shut down and delete the image) this server.
"""
self._delete("/servers/%s" % base.getid(server))
def share_ip(self, server, ipgroup, address, configure=True):
"""
Share an IP address from the given IP group onto a server.
:param server: The :class:`Server` (or its ID) to share onto.
:param ipgroup: The :class:`IPGroup` that the given address belongs to.
:param address: The IP address to share.
:param configure: If ``True``, the server will be automatically
configured to use this IP. I don't know why you'd
want this to be ``False``.
"""
server = base.getid(server)
ipgroup = base.getid(ipgroup)
body = {'shareIp': {'sharedIpGroupId': ipgroup,
'configureServer': configure}}
self._update("/servers/%s/ips/public/%s" % (server, address), body)
def unshare_ip(self, server, address):
"""
Stop sharing the given address.
:param server: The :class:`Server` (or its ID) to share onto.
:param address: The IP address to stop sharing.
"""
server = base.getid(server)
self._delete("/servers/%s/ips/public/%s" % (server, address))
def add_fixed_ip(self, server, network_id):
"""
Add an IP address on a network.
:param server: The :class:`Server` (or its ID) to add an IP to.
:param network_id: The ID of the network the IP should be on.
"""
self._action('addFixedIp', server, {'networkId': network_id})
def remove_fixed_ip(self, server, address):
"""
Remove an IP address.
:param server: The :class:`Server` (or its ID) to add an IP to.
:param address: The IP address to remove.
"""
self._action('removeFixedIp', server, {'address': address})
def reboot(self, server, type=REBOOT_SOFT):
"""
Reboot a server.
:param server: The :class:`Server` (or its ID) to share onto.
:param type: either :data:`REBOOT_SOFT` for a software-level reboot,
or `REBOOT_HARD` for a virtual power cycle hard reboot.
"""
self._action('reboot', server, {'type': type})
def rebuild(self, server, image):
"""
Rebuild -- shut down and then re-image -- a server.
:param server: The :class:`Server` (or its ID) to share onto.
:param image: the :class:`Image` (or its ID) to re-image with.
"""
self._action('rebuild', server, {'imageId': base.getid(image)})
def resize(self, server, flavor):
"""
Resize a server's resources.
:param server: The :class:`Server` (or its ID) to share onto.
:param flavor: the :class:`Flavor` (or its ID) to resize to.
Until a resize event is confirmed with :meth:`confirm_resize`, the old
server will be kept around and you'll be able to roll back to the old
flavor quickly with :meth:`revert_resize`. All resizes are
automatically confirmed after 24 hours.
"""
self._action('resize', server, {'flavorId': base.getid(flavor)})
def backup(self, server, image_name, backup_type, rotation):
"""
Create a server backup.
:param server: The :class:`Server` (or its ID).
:param image_name: The name to assign the newly create image.
:param backup_type: 'daily' or 'weekly'
:param rotation: number of backups of type 'backup_type' to keep
:returns Newly created :class:`Image` object
"""
if not rotation:
raise Exception("rotation is required for backups")
elif not backup_type:
raise Exception("backup_type required for backups")
elif backup_type not in ("daily", "weekly"):
raise Exception("Invalid backup_type: must be daily or weekly")
data = {
"name": image_name,
"rotation": rotation,
"backup_type": backup_type,
}
self._action('createBackup', server, data)
def pause(self, server):
"""
Pause the server.
"""
self.api.client.post('/servers/%s/pause' % base.getid(server))
def unpause(self, server):
"""
Unpause the server.
"""
self.api.client.post('/servers/%s/unpause' % base.getid(server))
def suspend(self, server):
"""
Suspend the server.
"""
self.api.client.post('/servers/%s/suspend' % base.getid(server))
def resume(self, server):
"""
Resume the server.
"""
self.api.client.post('/servers/%s/resume' % base.getid(server))
def rescue(self, server):
"""
Rescue the server.
"""
self.api.client.post('/servers/%s/rescue' % base.getid(server))
def unrescue(self, server):
"""
Unrescue the server.
"""
self.api.client.post('/servers/%s/unrescue' % base.getid(server))
def diagnostics(self, server):
"""Retrieve server diagnostics."""
return self.api.client.get("/servers/%s/diagnostics" %
base.getid(server))
def actions(self, server):
"""Retrieve server actions."""
return self._list("/servers/%s/actions" % base.getid(server),
"actions")
def confirm_resize(self, server):
"""
Confirm that the resize worked, thus removing the original server.
:param server: The :class:`Server` (or its ID) to share onto.
"""
self._action('confirmResize', server)
def revert_resize(self, server):
"""
Revert a previous resize, switching back to the old server.
:param server: The :class:`Server` (or its ID) to share onto.
"""
self._action('revertResize', server)
def migrate(self, server):
"""
Migrate a server to a new host in the same zone.
:param server: The :class:`Server` (or its ID).
"""
self.api.client.post('/servers/%s/migrate' % base.getid(server))
def _action(self, action, server, info=None):
"""
Perform a server "action" -- reboot/rebuild/resize/etc.
"""
self.api.client.post('/servers/%s/action' % base.getid(server),
body={action: info})

788
novaclient/v1_0/shell.py
View File

@ -1,788 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 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 getpass
import os
from novaclient import exceptions
from novaclient import utils
from novaclient.v1_0 import client
from novaclient.v1_0 import backup_schedules
from novaclient.v1_0 import servers
CLIENT_CLASS = client.Client
# Choices for flags.
DAY_CHOICES = [getattr(backup_schedules, i).lower()
for i in dir(backup_schedules)
if i.startswith('BACKUP_WEEKLY_')]
HOUR_CHOICES = [getattr(backup_schedules, i).lower()
for i in dir(backup_schedules)
if i.startswith('BACKUP_DAILY_')]
# Sentinal for boot --key
AUTO_KEY = object()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('--enable', dest='enabled', default=None, action='store_true',
help='Enable backups.')
@utils.arg('--disable', dest='enabled', action='store_false',
help='Disable backups.')
@utils.arg('--weekly', metavar='<day>', choices=DAY_CHOICES,
help='Schedule a weekly backup for <day> (one of: %s).' %
utils.pretty_choice_list(DAY_CHOICES))
@utils.arg('--daily', metavar='<time-window>', choices=HOUR_CHOICES,
help='Schedule a daily backup during <time-window> (one of: %s).' %
utils.pretty_choice_list(HOUR_CHOICES))
def do_backup_schedule(cs, args):
"""
Show or edit the backup schedule for a server.
With no flags, the backup schedule will be shown. If flags are given,
the backup schedule will be modified accordingly.
"""
server = _find_server(cs, args.server)
# If we have some flags, update the backup
backup = {}
if args.daily:
backup['daily'] = getattr(backup_schedules, 'BACKUP_DAILY_%s' %
args.daily.upper())
if args.weekly:
backup['weekly'] = getattr(backup_schedules, 'BACKUP_WEEKLY_%s' %
args.weekly.upper())
if args.enabled is not None:
backup['enabled'] = args.enabled
if backup:
server.backup_schedule.update(**backup)
else:
utils.print_dict(server.backup_schedule._info)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_backup_schedule_delete(cs, args):
"""
Delete the backup schedule for a server.
"""
server = _find_server(cs, args.server)
server.backup_schedule.delete()
def _boot(cs, args, reservation_id=None, min_count=None, max_count=None):
"""Boot a new server."""
if min_count is None:
min_count = 1
if max_count is None:
max_count = min_count
if min_count > max_count:
raise exceptions.CommandError("min_instances should be"
"<= max_instances")
if not min_count or not max_count:
raise exceptions.CommandError("min_instances nor max_instances"
"should be 0")
flavor = args.flavor or cs.flavors.find(ram=256)
image = args.image or cs.images.find(name="Ubuntu 10.04 LTS "\
"(lucid)")
# Map --ipgroup <name> to an ID.
# XXX do this for flavor/image?
if args.ipgroup:
ipgroup = _find_ipgroup(cs, args.ipgroup)
else:
ipgroup = None
metadata = dict(v.split('=') for v in args.meta)
files = {}
for f in args.files:
dst, src = f.split('=', 1)
try:
files[dst] = open(src)
except IOError, e:
raise exceptions.CommandError("Can't open '%s': %s" % (src, e))
if args.key is AUTO_KEY:
possible_keys = [os.path.join(os.path.expanduser('~'), '.ssh', k)
for k in ('id_dsa.pub', 'id_rsa.pub')]
for k in possible_keys:
if os.path.exists(k):
keyfile = k
break
else:
raise exceptions.CommandError("Couldn't find a key file: tried "
"~/.ssh/id_dsa.pub or ~/.ssh/id_rsa.pub")
elif args.key:
keyfile = args.key
else:
keyfile = None
if keyfile:
try:
files['/root/.ssh/authorized_keys2'] = open(keyfile)
except IOError, e:
raise exceptions.CommandError("Can't open '%s': %s" % (keyfile, e))
return (args.name, image, flavor, ipgroup, metadata, files,
reservation_id, min_count, max_count)
@utils.arg('--flavor',
default=None,
type=int,
metavar='<flavor>',
help="Flavor ID (see 'nova flavors'). "\
"Defaults to 256MB RAM instance.")
@utils.arg('--image',
default=None,
type=int,
metavar='<image>',
help="Image ID (see 'nova images'). "\
"Defaults to Ubuntu 10.04 LTS.")
@utils.arg('--ipgroup',
default=None,
metavar='<group>',
help="IP group name or ID (see 'nova ipgroup-list').")
@utils.arg('--meta',
metavar="<key=value>",
action='append',
default=[],
help="Record arbitrary key/value metadata. "\
"May be give multiple times.")
@utils.arg('--file',
metavar="<dst-path=src-path>",
action='append',
dest='files',
default=[],
help="Store arbitrary files from <src-path> locally to <dst-path> "\
"on the new server. You may store up to 5 files.")
@utils.arg('--key',
metavar='<path>',
nargs='?',
const=AUTO_KEY,
help="Key the server with an SSH keypair. "\
"Looks in ~/.ssh for a key, "\
"or takes an explicit <path> to one.")
@utils.arg('name', metavar='<name>', help='Name for the new server')
def do_boot(cs, args):
"""Boot a new server."""
name, image, flavor, ipgroup, metadata, files, reservation_id, \
min_count, max_count = _boot(cs, args)
server = cs.servers.create(args.name, image, flavor,
ipgroup=ipgroup,
meta=metadata,
files=files,
min_count=min_count,
max_count=max_count)
utils.print_dict(server._info)
@utils.arg('--flavor',
default=None,
type=int,
metavar='<flavor>',
help="Flavor ID (see 'nova flavors'). "\
"Defaults to 256MB RAM instance.")
@utils.arg('--image',
default=None,
type=int,
metavar='<image>',
help="Image ID (see 'nova images'). "\
"Defaults to Ubuntu 10.04 LTS.")
@utils.arg('--ipgroup',
default=None,
metavar='<group>',
help="IP group name or ID (see 'nova ipgroup-list').")
@utils.arg('--meta',
metavar="<key=value>",
action='append',
default=[],
help="Record arbitrary key/value metadata. "\
"May be give multiple times.")
@utils.arg('--file',
metavar="<dst-path=src-path>",
action='append',
dest='files',
default=[],
help="Store arbitrary files from <src-path> locally to <dst-path> "\
"on the new server. You may store up to 5 files.")
@utils.arg('--key',
metavar='<path>',
nargs='?',
const=AUTO_KEY,
help="Key the server with an SSH keypair. "\
"Looks in ~/.ssh for a key, "\
"or takes an explicit <path> to one.")
@utils.arg('account', metavar='<account>', help='Account to build this'\
' server for')
@utils.arg('name', metavar='<name>', help='Name for the new server')
def do_boot_for_account(cs, args):
"""Boot a new server in an account."""
name, image, flavor, ipgroup, metadata, files, reservation_id, \
min_count, max_count = _boot(cs, args)
server = cs.accounts.create_instance_for(args.account, args.name,
image, flavor,
ipgroup=ipgroup,
meta=metadata,
files=files)
utils.print_dict(server._info)
@utils.arg('--flavor',
default=None,
type=int,
metavar='<flavor>',
help="Flavor ID (see 'nova flavors'). "\
"Defaults to 256MB RAM instance.")
@utils.arg('--image',
default=None,
type=int,
metavar='<image>',
help="Image ID (see 'nova images'). "\
"Defaults to Ubuntu 10.04 LTS.")
@utils.arg('--ipgroup',
default=None,
metavar='<group>',
help="IP group name or ID (see 'nova ipgroup-list').")
@utils.arg('--meta',
metavar="<key=value>",
action='append',
default=[],
help="Record arbitrary key/value metadata. "\
"May be give multiple times.")
@utils.arg('--file',
metavar="<dst-path=src-path>",
action='append',
dest='files',
default=[],
help="Store arbitrary files from <src-path> locally to <dst-path> "\
"on the new server. You may store up to 5 files.")
@utils.arg('--key',
metavar='<path>',
nargs='?',
const=AUTO_KEY,
help="Key the server with an SSH keypair. "\
"Looks in ~/.ssh for a key, "\
"or takes an explicit <path> to one.")
@utils.arg('--reservation_id',
default=None,
metavar='<reservation_id>',
help="Reservation ID (a UUID). "\
"If unspecified will be generated by the server.")
@utils.arg('--min_instances',
default=None,
type=int,
metavar='<number>',
help="The minimum number of instances to build. "\
"Defaults to 1.")
@utils.arg('--max_instances',
default=None,
type=int,
metavar='<number>',
help="The maximum number of instances to build. "\
"Defaults to 'min_instances' setting.")
@utils.arg('name', metavar='<name>', help='Name for the new server')
def do_zone_boot(cs, args):
"""Boot a new server, potentially across Zones."""
reservation_id = args.reservation_id
min_count = args.min_instances
max_count = args.max_instances
name, image, flavor, ipgroup, metadata, \
files, reservation_id, min_count, max_count = \
_boot(cs, args,
reservation_id=reservation_id,
min_count=min_count,
max_count=max_count)
reservation_id = cs.zones.boot(args.name, image, flavor,
ipgroup=ipgroup,
meta=metadata,
files=files,
reservation_id=reservation_id,
min_count=min_count,
max_count=max_count)
print "Reservation ID=", reservation_id
def _translate_flavor_keys(collection):
convert = [('ram', 'memory_mb'), ('disk', 'local_gb')]
for item in collection:
keys = item.__dict__.keys()
for from_key, to_key in convert:
if from_key in keys and to_key not in keys:
setattr(item, to_key, item._info[from_key])
def do_flavor_list(cs, args):
"""Print a list of available 'flavors' (sizes of servers)."""
flavors = cs.flavors.list()
_translate_flavor_keys(flavors)
utils.print_list(flavors, [
'ID',
'Name',
'Memory_MB',
'Swap',
'Local_GB',
'VCPUs',
'RXTX_Factor'])
def do_image_list(cs, args):
"""Print a list of available images to boot from."""
server_list = {}
for server in cs.servers.list():
server_list[server.id] = server.name
image_list = cs.images.list()
for i in range(len(image_list)):
if hasattr(image_list[i], 'serverId'):
image_list[i].serverId = server_list[image_list[i].serverId] + \
' (' + str(image_list[i].serverId) + ')'
utils.print_list(image_list, ['ID', 'Name', 'serverId', 'Status'])
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('name', metavar='<name>', help='Name of snapshot.')
def do_image_create(cs, args):
"""Create a new image by taking a snapshot of a running server."""
server = _find_server(cs, args.server)
image = cs.images.create(server, args.name)
utils.print_dict(image._info)
@utils.arg('image', metavar='<image>', help='Name or ID of image.')
def do_image_delete(cs, args):
"""
Delete an image.
It should go without saying, but you can only delete images you
created.
"""
image = _find_image(cs, args.image)
image.delete()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('group', metavar='<group>', help='Name or ID of group.')
@utils.arg('address', metavar='<address>', help='IP address to share.')
def do_ip_share(cs, args):
"""Share an IP address from the given IP group onto a server."""
server = _find_server(cs, args.server)
group = _find_ipgroup(cs, args.group)
server.share_ip(group, args.address)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('address', metavar='<address>',
help='Shared IP address to remove from the server.')
def do_ip_unshare(cs, args):
"""Stop sharing an given address with a server."""
server = _find_server(cs, args.server)
server.unshare_ip(args.address)
def do_ipgroup_list(cs, args):
"""Show IP groups."""
def pretty_server_list(ipgroup):
return ", ".join(cs.servers.get(id).name
for id in ipgroup.servers)
utils.print_list(cs.ipgroups.list(),
fields=['ID', 'Name', 'Server List'],
formatters={'Server List': pretty_server_list})
@utils.arg('group', metavar='<group>', help='Name or ID of group.')
def do_ipgroup_show(cs, args):
"""Show details about a particular IP group."""
group = _find_ipgroup(cs, args.group)
utils.print_dict(group._info)
@utils.arg('name', metavar='<name>', help='What to name this new group.')
@utils.arg('server', metavar='<server>', nargs='?',
help='Server (name or ID) to make a member of this new group.')
def do_ipgroup_create(cs, args):
"""Create a new IP group."""
if args.server:
server = _find_server(cs, args.server)
else:
server = None
group = cs.ipgroups.create(args.name, server)
utils.print_dict(group._info)
@utils.arg('group', metavar='<group>', help='Name or ID of group.')
def do_ipgroup_delete(cs, args):
"""Delete an IP group."""
_find_ipgroup(cs, args.group).delete()
@utils.arg('--fixed_ip',
dest='fixed_ip',
metavar='<fixed_ip>',
default=None,
help='Only match against fixed IP.')
@utils.arg('--reservation_id',
dest='reservation_id',
metavar='<reservation_id>',
default=None,
help='Only return instances that match reservation_id.')
@utils.arg('--recurse_zones',
dest='recurse_zones',
metavar='<0|1>',
nargs='?',
type=int,
const=1,
default=0,
help='Recurse through all zones if set.')
@utils.arg('--ip',
dest='ip',
metavar='<ip_regexp>',
default=None,
help='Search with regular expression match by IP address')
@utils.arg('--ip6',
dest='ip6',
metavar='<ip6_regexp>',
default=None,
help='Search with regular expression match by IPv6 address')
@utils.arg('--name',
dest='name',
metavar='<name_regexp>',
default=None,
help='Search with regular expression match by name')
@utils.arg('--instance_name',
dest='instance_name',
metavar='<name_regexp>',
default=None,
help='Search with regular expression match by instance name')
@utils.arg('--status',
dest='status',
metavar='<status>',
default=None,
help='Search by server status')
@utils.arg('--flavor',
dest='flavor',
metavar='<flavor>',
type=int,
default=None,
help='Search by flavor ID')
@utils.arg('--image',
dest='image',
type=int,
metavar='<image>',
default=None,
help='Search by image ID')
@utils.arg('--host',
dest='host',
metavar='<hostname>',
default=None,
help="Search by instances by hostname to which they are assigned")
def do_list(cs, args):
"""List active servers."""
recurse_zones = args.recurse_zones
search_opts = {
'reservation_id': args.reservation_id,
'fixed_ip': args.fixed_ip,
'recurse_zones': recurse_zones,
'ip': args.ip,
'ip6': args.ip6,
'name': args.name,
'image': args.image,
'flavor': args.flavor,
'status': args.status,
'host': args.host,
'instance_name': args.instance_name}
if recurse_zones:
to_print = ['UUID', 'Name', 'Status', 'Public IP', 'Private IP']
else:
to_print = ['ID', 'Name', 'Status', 'Public IP', 'Private IP']
utils.print_list(cs.servers.list(search_opts=search_opts),
to_print)
@utils.arg('--hard',
dest='reboot_type',
action='store_const',
const=servers.REBOOT_HARD,
default=servers.REBOOT_SOFT,
help='Perform a hard reboot (instead of a soft one).')
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_reboot(cs, args):
"""Reboot a server."""
_find_server(cs, args.server).reboot(args.reboot_type)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('image', metavar='<image>', help="Name or ID of new image.")
def do_rebuild(cs, args):
"""Shutdown, re-image, and re-boot a server."""
server = _find_server(cs, args.server)
image = _find_image(cs, args.image)
server.rebuild(image)
@utils.arg('server', metavar='<server>',
help='Name (old name) or ID of server.')
@utils.arg('name', metavar='<name>', help='New name for the server.')
def do_rename(cs, args):
"""Rename a server."""
_find_server(cs, args.server).update(name=args.name)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('flavor', metavar='<flavor>', help="Name or ID of new flavor.")
def do_resize(cs, args):
"""Resize a server."""
server = _find_server(cs, args.server)
flavor = _find_flavor(cs, args.flavor)
server.resize(flavor)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('name', metavar='<name>', help='Name of snapshot.')
@utils.arg('backup_type', metavar='<daily|weekly>', help='type of backup')
@utils.arg('rotation', type=int, metavar='<rotation>',
help="Number of backups to retain. Used for backup image_type.")
def do_backup(cs, args):
"""Backup a server."""
server = _find_server(cs, args.server)
server.backup(args.name, args.backup_type, args.rotation)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_migrate(cs, args):
"""Migrate a server."""
_find_server(cs, args.server).migrate()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_pause(cs, args):
"""Pause a server."""
_find_server(cs, args.server).pause()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_unpause(cs, args):
"""Unpause a server."""
_find_server(cs, args.server).unpause()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_suspend(cs, args):
"""Suspend a server."""
_find_server(cs, args.server).suspend()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_resume(cs, args):
"""Resume a server."""
_find_server(cs, args.server).resume()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_rescue(cs, args):
"""Rescue a server."""
_find_server(cs, args.server).rescue()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_unrescue(cs, args):
"""Unrescue a server."""
_find_server(cs, args.server).unrescue()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_diagnostics(cs, args):
"""Retrieve server diagnostics."""
utils.print_dict(cs.servers.diagnostics(args.server)[1])
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_actions(cs, args):
"""Retrieve server actions."""
utils.print_list(
cs.servers.actions(args.server),
["Created_At", "Action", "Error"])
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_resize_confirm(cs, args):
"""Confirm a previous resize."""
_find_server(cs, args.server).confirm_resize()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_resize_revert(cs, args):
"""Revert a previous resize (and return to the previous VM)."""
_find_server(cs, args.server).revert_resize()
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_root_password(cs, args):
"""
Change the root password for a server.
"""
server = _find_server(cs, args.server)
p1 = getpass.getpass('New password: ')
p2 = getpass.getpass('Again: ')
if p1 != p2:
raise exceptions.CommandError("Passwords do not match.")
server.update(password=p1)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_show(cs, args):
"""Show details about the given server."""
s = _find_server(cs, args.server)
info = s._info.copy()
addresses = info.pop('addresses')
for addrtype in addresses:
info['%s ip' % addrtype] = ', '.join(addresses[addrtype])
flavorId = info.get('flavorId', None)
if flavorId:
info['flavor'] = _find_flavor(cs, info.pop('flavorId')).name
imageId = info.get('imageId', None)
if imageId:
info['image'] = _find_image(cs, info.pop('imageId')).name
utils.print_dict(info)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
def do_delete(cs, args):
"""Immediately shut down and delete a server."""
_find_server(cs, args.server).delete()
# --zone_username is required since --username is already used.
@utils.arg('zone', metavar='<zone_id>', help='ID of the zone', default=None)
@utils.arg('--api_url', dest='api_url', default=None, help='New URL.')
@utils.arg('--zone_username', dest='zone_username', default=None,
help='New zone username.')
@utils.arg('--zone_password', dest='zone_password', default=None,
help='New password.')
@utils.arg('--weight_offset', dest='weight_offset', default=None,
help='Child Zone weight offset.')
@utils.arg('--weight_scale', dest='weight_scale', default=None,
help='Child Zone weight scale.')
def do_zone(cs, args):
"""Show or edit a child zone. No zone arg for this zone."""
zone = cs.zones.get(args.zone)
# If we have some flags, update the zone
zone_delta = {}
if args.api_url:
zone_delta['api_url'] = args.api_url
if args.zone_username:
zone_delta['username'] = args.zone_username
if args.zone_password:
zone_delta['password'] = args.zone_password
if args.weight_offset:
zone_delta['weight_offset'] = args.weight_offset
if args.weight_scale:
zone_delta['weight_scale'] = args.weight_scale
if zone_delta:
zone.update(**zone_delta)
else:
utils.print_dict(zone._info)
def do_zone_info(cs, args):
"""Get this zones name and capabilities."""
zone = cs.zones.info()
utils.print_dict(zone._info)
@utils.arg('zone_name', metavar='<zone_name>',
help='Name of the child zone being added.')
@utils.arg('api_url', metavar='<api_url>', help="URL for the Zone's Auth API")
@utils.arg('--zone_username', metavar='<zone_username>',
help='Optional Authentication username. (Default=None)',
default=None)
@utils.arg('--zone_password', metavar='<zone_password>',
help='Authentication password. (Default=None)',
default=None)
@utils.arg('--weight_offset', metavar='<weight_offset>',
help='Child Zone weight offset (Default=0.0))',
default=0.0)
@utils.arg('--weight_scale', metavar='<weight_scale>',
help='Child Zone weight scale (Default=1.0).',
default=1.0)
def do_zone_add(cs, args):
"""Add a new child zone."""
zone = cs.zones.create(args.zone_name, args.api_url,
args.zone_username, args.zone_password,
args.weight_offset, args.weight_scale)
utils.print_dict(zone._info)
@utils.arg('zone', metavar='<zone>', help='Name or ID of the zone')
def do_zone_delete(cs, args):
"""Delete a zone."""
cs.zones.delete(args.zone)
def do_zone_list(cs, args):
"""List the children of a zone."""
utils.print_list(cs.zones.list(), ['ID', 'Name', 'Is Active', \
'API URL', 'Weight Offset', 'Weight Scale'])
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('network_id', metavar='<network_id>', help='Network ID.')
def do_add_fixed_ip(cs, args):
"""Add new IP address to network."""
server = _find_server(cs, args.server)
server.add_fixed_ip(args.network_id)
@utils.arg('server', metavar='<server>', help='Name or ID of server.')
@utils.arg('address', metavar='<address>', help='IP Address.')
def do_remove_fixed_ip(cs, args):
"""Remove an IP address from a server."""
server = _find_server(cs, args.server)
server.remove_fixed_ip(args.address)
def _find_server(cs, server):
"""Get a server by name or ID."""
return utils.find_resource(cs.servers, server)
def _find_ipgroup(cs, group):
"""Get an IP group by name or ID."""
return utils.find_resource(cs.ipgroups, group)
def _find_image(cs, image):
"""Get an image by name or ID."""
return utils.find_resource(cs.images, image)
def _find_flavor(cs, flavor):
"""Get a flavor by name, ID, or RAM size."""
try:
return utils.find_resource(cs.flavors, flavor)
except exceptions.NotFound:
return cs.flavors.find(ram=flavor)

199
novaclient/v1_0/zones.py
View File

@ -1,199 +0,0 @@
# Copyright 2011 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.
"""
Zone interface.
"""
from novaclient import base
from novaclient.v1_0 import base as local_base
class Weighting(base.Resource):
def __init__(self, manager, info, loaded=False):
self.name = "n/a"
super(Weighting, self).__init__(manager, info, loaded)
def __repr__(self):
return "<Weighting: %s>" % self.name
def to_dict(self):
"""Return the original info setting, which is a dict."""
return self._info
class Zone(base.Resource):
def __init__(self, manager, info, loaded=False):
self.name = "n/a"
self.is_active = "n/a"
self.capabilities = "n/a"
super(Zone, self).__init__(manager, info, loaded)
def __repr__(self):
return "<Zone: %s>" % self.api_url
def delete(self):
"""
Delete a child zone.
"""
self.manager.delete(self)
def update(self, api_url=None, username=None, password=None,
weight_offset=None, weight_scale=None):
"""
Update the name for this child zone.
:param api_url: Update the child zone's API URL.
:param username: Update the child zone's username.
:param password: Update the child zone's password.
:param weight_offset: Update the child zone's weight offset.
:param weight_scale: Update the child zone's weight scale.
"""
self.manager.update(self, api_url, username, password,
weight_offset, weight_scale)
class ZoneManager(local_base.BootingManagerWithFind):
resource_class = Zone
def info(self):
"""
Get info on this zone.
:rtype: :class:`Zone`
"""
return self._get("/zones/info", "zone")
def get(self, zone):
"""
Get a child zone.
:param server: ID of the :class:`Zone` to get.
:rtype: :class:`Zone`
"""
return self._get("/zones/%s" % base.getid(zone), "zone")
def list(self, detailed=True):
"""
Get a list of child zones.
:rtype: list of :class:`Zone`
"""
detail = ""
if detailed:
detail = "/detail"
return self._list("/zones%s" % detail, "zones")
def create(self, zone_name, api_url, username, password,
weight_offset=0.0, weight_scale=1.0):
"""
Create a new child zone.
:param zone_name: The child zone's name.
:param api_url: The child zone's auth URL.
:param username: The child zone's username.
:param password: The child zone's password.
:param weight_offset: The child zone's weight offset.
:param weight_scale: The child zone's weight scale.
"""
body = {"zone": {
"name": zone_name,
"api_url": api_url,
"username": username,
"password": password,
"weight_offset": weight_offset,
"weight_scale": weight_scale
}}
return self._create("/zones", body, "zone")
def boot(self, name, image, flavor, ipgroup=None, meta=None, files=None,
zone_blob=None, reservation_id=None, min_count=None,
max_count=None):
"""
Create (boot) a new server while being aware of Zones.
:param name: Something to name the server.
:param image: The :class:`Image` to boot with.
:param flavor: The :class:`Flavor` to boot onto.
:param ipgroup: An initial :class:`IPGroup` for this server.
:param meta: A dict of arbitrary key/value metadata to store for this
server. A maximum of five entries is allowed, and both
keys and values must be 255 characters or less.
:param files: A dict of files to overrwrite on the server upon boot.
Keys are file names (i.e. ``/etc/passwd``) and values
are the file contents (either as a string or as a
file-like object). A maximum of five entries is allowed,
and each file must be 10k or less.
:param zone_blob: a single (encrypted) string which is used internally
by Nova for routing between Zones. Users cannot populate
this field.
:param reservation_id: a UUID for the set of servers being requested.
:param min_count: minimum number of servers to create.
:param max_count: maximum number of servers to create.
"""
if not min_count:
min_count = 1
if not max_count:
max_count = min_count
return self._boot("/zones/boot", "reservation_id", name, image, flavor,
ipgroup=ipgroup, meta=meta, files=files,
zone_blob=zone_blob, reservation_id=reservation_id,
return_raw=True, min_count=min_count,
max_count=max_count)
def select(self, *args, **kwargs):
"""
Given requirements for a new instance, select hosts
in this zone that best match those requirements.
"""
# 'specs' may be passed in as None, so change to an empty string.
specs = kwargs.get("specs") or ""
url = "/zones/select"
weighting_list = self._list(url, "weights", Weighting, body=specs)
return [wt.to_dict() for wt in weighting_list]
def delete(self, zone):
"""
Delete a child zone.
"""
self._delete("/zones/%s" % base.getid(zone))
def update(self, zone, api_url=None, username=None, password=None,
weight_offset=None, weight_scale=None):
"""
Update the name or the api_url for a zone.
:param zone: The :class:`Zone` (or its ID) to update.
:param api_url: Update the API URL.
:param username: Update the username.
:param password: Update the password.
:param weight_offset: Update the child zone's weight offset.
:param weight_scale: Update the child zone's weight scale.
"""
body = {"zone": {}}
if api_url:
body["zone"]["api_url"] = api_url
if username:
body["zone"]["username"] = username
if password:
body["zone"]["password"] = password
if weight_offset:
body["zone"]["weight_offset"] = weight_offset
if weight_scale:
body["zone"]["weight_scale"] = weight_scale
if not len(body["zone"]):
return
self._update("/zones/%s" % base.getid(zone), body)

4
tests/test_base.py
View File

@ -1,8 +1,8 @@
from novaclient import base
from novaclient import exceptions
from novaclient.v1_0 import flavors
from novaclient.v1_1 import flavors
from tests import utils
from tests.v1_0 import fakes
from tests.v1_1 import fakes
cs = fakes.FakeClient()

0
tests/v1_0/__init__.py
View File

418
tests/v1_0/fakes.py
View File

@ -1,418 +0,0 @@
import httplib2
from novaclient import client as base_client
from novaclient.v1_0 import client
from tests import fakes
class FakeClient(fakes.FakeClient, client.Client):
def __init__(self, *args, **kwargs):
client.Client.__init__(self, 'username', 'password',
'project_id', 'auth_url')
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
munged_url = url.strip('/').replace('/', '_').replace('.', '_')
callback = "%s_%s" % (method.lower(), munged_url)
if not hasattr(self, callback):
raise AssertionError('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 body.keys() == ['server']
fakes.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']:
fakes.assert_has_keys(pfile, required=['path', 'contents'])
return (202, self.get_servers_1234()[1])
def post_servers_1234_migrate(self, *args, **kwargs):
return (202, None)
def post_servers_1234_rescue(self, *args, **kwargs):
return (202, None)
def post_servers_1234_unrescue(self, *args, **kwargs):
return (202, None)
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 body.keys() == ['server']
fakes.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 body.keys() == ['shareIp']
fakes.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 len(body.keys()) == 1
action = body.keys()[0]
if action == 'reboot':
assert body[action].keys() == ['type']
assert body[action]['type'] in ['HARD', 'SOFT']
elif action == 'rebuild':
assert body[action].keys() == ['imageId']
elif action == 'resize':
assert body[action].keys() == ['flavorId']
elif action == 'createBackup':
assert set(body[action].keys()) == \
set(['name', 'rotation', 'backup_type'])
elif action == 'confirmResize':
assert body[action] is None
# This one method returns a different response code
return (204, None)
elif action == 'revertResize':
assert body[action] is None
elif action == 'migrate':
assert body[action] is None
elif action == 'addFixedIp':
assert body[action].keys() == ['networkId']
elif action == 'removeFixedIp':
assert body[action].keys() == ['address']
else:
raise AssertionError("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": 1234,
"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 body.keys() == ['image']
fakes.assert_has_keys(body['image'], required=['serverId', 'name'])
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 body.keys() == ['backupSchedule']
fakes.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 body.keys() == ['sharedIpGroup']
fakes.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 body.keys() == ['zone']
fakes.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 body.keys() == ['zone']
fakes.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 body.keys() == ['server']
fakes.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']:
fakes.assert_has_keys(pfile, required=['path', 'contents'])
return (202, self.get_servers_1234()[1])

24
tests/v1_0/test_accounts.py
View File

@ -1,24 +0,0 @@
import StringIO
from tests import utils
from tests.v1_0 import fakes
cs = fakes.FakeClient()
class AccountsTest(utils.TestCase):
def test_instance_creation_for_account(self):
cs.accounts.create_instance_for(
account_id='test_account',
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
ipgroup=1,
files={
'/etc/passwd': 'some data', # a file
'/tmp/foo.txt': StringIO.StringIO('data') # a stream
})
cs.assert_called('POST', '/accounts/test_account/create_instance')

73
tests/v1_0/test_auth.py
View File

@ -1,73 +0,0 @@
import httplib2
import mock
from novaclient import exceptions
from novaclient.v1_0 import client
from tests import utils
class AuthenticationTests(utils.TestCase):
def test_authenticate_success(self):
cs = client.Client("username", "password", "project_id")
management_url = 'https://servers.api.rackspacecloud.com/v1.0/443470'
auth_response = httplib2.Response({
'status': 204,
'x-server-management-url': management_url,
'x-auth-token': '1b751d74-de0c-46ae-84f0-915744b582d1',
})
mock_request = mock.Mock(return_value=(auth_response, None))
@mock.patch.object(httplib2.Http, "request", mock_request)
def test_auth_call():
cs.client.authenticate()
headers = {
'X-Auth-User': 'username',
'X-Auth-Key': 'password',
'X-Auth-Project-Id': 'project_id',
'User-Agent': cs.client.USER_AGENT
}
mock_request.assert_called_with(cs.client.auth_url, 'GET',
headers=headers)
self.assertEqual(cs.client.management_url,
auth_response['x-server-management-url'])
self.assertEqual(cs.client.auth_token,
auth_response['x-auth-token'])
test_auth_call()
def test_authenticate_failure(self):
cs = client.Client("username", "password", "project_id")
auth_response = httplib2.Response({'status': 401})
mock_request = mock.Mock(return_value=(auth_response, None))
@mock.patch.object(httplib2.Http, "request", mock_request)
def test_auth_call():
self.assertRaises(exceptions.Unauthorized, cs.client.authenticate)
test_auth_call()
def test_auth_automatic(self):
cs = client.Client("username", "password", "project_id")
http_client = cs.client
http_client.management_url = ''
mock_request = mock.Mock(return_value=(None, None))
@mock.patch.object(http_client, 'request', mock_request)
@mock.patch.object(http_client, 'authenticate')
def test_auth_call(m):
http_client.get('/')
m.assert_called()
mock_request.assert_called()
test_auth_call()
def test_auth_manual(self):
cs = client.Client("username", "password", "project_id")
@mock.patch.object(cs.client, 'authenticate')
def test_auth_call(m):
cs.authenticate()
m.assert_called()
test_auth_call()

59
tests/v1_0/test_backup_schedules.py
View File

@ -1,59 +0,0 @@
from novaclient.v1_0 import backup_schedules
from tests import utils
from tests.v1_0 import fakes
cs = fakes.FakeClient()
class BackupSchedulesTest(utils.TestCase):
def test_get_backup_schedule(self):
s = cs.servers.get(1234)
# access via manager
b = cs.backup_schedules.get(server=s)
self.assertTrue(isinstance(b, backup_schedules.BackupSchedule))
cs.assert_called('GET', '/servers/1234/backup_schedule')
b = cs.backup_schedules.get(server=1234)
self.assertTrue(isinstance(b, backup_schedules.BackupSchedule))
cs.assert_called('GET', '/servers/1234/backup_schedule')
# access via instance
self.assertTrue(isinstance(s.backup_schedule,
backup_schedules.BackupSchedule))
cs.assert_called('GET', '/servers/1234/backup_schedule')
# Just for coverage's sake
b = s.backup_schedule.get()
cs.assert_called('GET', '/servers/1234/backup_schedule')
def test_create_update_backup_schedule(self):
s = cs.servers.get(1234)
# create/update via manager
cs.backup_schedules.update(
server=s,
enabled=True,
weekly=backup_schedules.BACKUP_WEEKLY_THURSDAY,
daily=backup_schedules.BACKUP_DAILY_H_1000_1200
)
cs.assert_called('POST', '/servers/1234/backup_schedule')
# and via instance
s.backup_schedule.update(enabled=False)
cs.assert_called('POST', '/servers/1234/backup_schedule')
def test_delete_backup_schedule(self):
s = cs.servers.get(1234)
# delete via manager
cs.backup_schedules.delete(s)
cs.assert_called('DELETE', '/servers/1234/backup_schedule')
cs.backup_schedules.delete(1234)
cs.assert_called('DELETE', '/servers/1234/backup_schedule')
# and via instance
s.backup_schedule.delete()
cs.assert_called('DELETE', '/servers/1234/backup_schedule')

37
tests/v1_0/test_flavors.py
View File

@ -1,37 +0,0 @@
from novaclient import exceptions
from novaclient.v1_0 import flavors
from tests import utils
from tests.v1_0 import fakes
cs = fakes.FakeClient()
class FlavorsTest(utils.TestCase):
def test_list_flavors(self):
fl = cs.flavors.list()
cs.assert_called('GET', '/flavors/detail')
[self.assertTrue(isinstance(f, flavors.Flavor)) for f in fl]
def test_list_flavors_undetailed(self):
fl = cs.flavors.list(detailed=False)
cs.assert_called('GET', '/flavors')
[self.assertTrue(isinstance(f, flavors.Flavor)) for f in fl]
def test_get_flavor_details(self):
f = cs.flavors.get(1)
cs.assert_called('GET', '/flavors/1')
self.assertTrue(isinstance(f, flavors.Flavor))
self.assertEqual(f.ram, 256)
self.assertEqual(f.disk, 10)
def test_find(self):
f = cs.flavors.find(ram=256)
cs.assert_called('GET', '/flavors/detail')
self.assertEqual(f.name, '256 MB Server')
f = cs.flavors.find(disk=20)
self.assertEqual(f.name, '512 MB Server')
self.assertRaises(exceptions.NotFound, cs.flavors.find, disk=12345)

44
tests/v1_0/test_images.py
View File

@ -1,44 +0,0 @@
from novaclient.v1_0 import images
from tests import utils
from tests.v1_0 import fakes
cs = fakes.FakeClient()
class ImagesTest(utils.TestCase):
def test_list_images(self):
il = cs.images.list()
cs.assert_called('GET', '/images/detail')
[self.assertTrue(isinstance(i, images.Image)) for i in il]
def test_list_images_undetailed(self):
il = cs.images.list(detailed=False)
cs.assert_called('GET', '/images')
[self.assertTrue(isinstance(i, images.Image)) for i in il]
def test_get_image_details(self):
i = cs.images.get(1)
cs.assert_called('GET', '/images/1')
self.assertTrue(isinstance(i, images.Image))
self.assertEqual(i.id, 1)
self.assertEqual(i.name, 'CentOS 5.2')
def test_create_image(self):
i = cs.images.create(server=1234, name="Just in case")
cs.assert_called('POST', '/images')
self.assertTrue(isinstance(i, images.Image))
def test_delete_image(self):
cs.images.delete(1)
cs.assert_called('DELETE', '/images/1')
def test_find(self):
i = cs.images.find(name="CentOS 5.2")
self.assertEqual(i.id, 1)
cs.assert_called('GET', '/images/detail')
iml = cs.images.findall(status='SAVING')
self.assertEqual(len(iml), 1)
self.assertEqual(iml[0].name, 'My Server Backup')

47
tests/v1_0/test_ipgroups.py
View File

@ -1,47 +0,0 @@
from novaclient.v1_0 import ipgroups
from tests import utils
from tests.v1_0 import fakes
cs = fakes.FakeClient()
class IPGroupTest(utils.TestCase):
def test_list_ipgroups(self):
ipl = cs.ipgroups.list()
cs.assert_called('GET', '/shared_ip_groups/detail')
[self.assertTrue(isinstance(ipg, ipgroups.IPGroup)) \
for ipg in ipl]
def test_list_ipgroups_undetailed(self):
ipl = cs.ipgroups.list(detailed=False)
cs.assert_called('GET', '/shared_ip_groups')
[self.assertTrue(isinstance(ipg, ipgroups.IPGroup)) \
for ipg in ipl]
def test_get_ipgroup(self):
ipg = cs.ipgroups.get(1)
cs.assert_called('GET', '/shared_ip_groups/1')
self.assertTrue(isinstance(ipg, ipgroups.IPGroup))
def test_create_ipgroup(self):
ipg = cs.ipgroups.create("My group", 1234)
cs.assert_called('POST', '/shared_ip_groups')
self.assertTrue(isinstance(ipg, ipgroups.IPGroup))
def test_delete_ipgroup(self):
ipg = cs.ipgroups.get(1)
ipg.delete()
cs.assert_called('DELETE', '/shared_ip_groups/1')
cs.ipgroups.delete(ipg)
cs.assert_called('DELETE', '/shared_ip_groups/1')
cs.ipgroups.delete(1)
cs.assert_called('DELETE', '/shared_ip_groups/1')
def test_find(self):
ipg = cs.ipgroups.find(name='group1')
cs.assert_called('GET', '/shared_ip_groups/detail')
self.assertEqual(ipg.name, 'group1')
ipgl = cs.ipgroups.findall(id=1)
self.assertEqual(ipgl, [ipgroups.IPGroup(None, {'id': 1})])

182
tests/v1_0/test_servers.py
View File

@ -1,182 +0,0 @@
import StringIO
from novaclient.v1_0 import servers
from tests import utils
from tests.v1_0 import fakes
cs = fakes.FakeClient()
class ServersTest(utils.TestCase):
def test_list_servers(self):
sl = cs.servers.list()
cs.assert_called('GET', '/servers/detail')
[self.assertTrue(isinstance(s, servers.Server)) for s in sl]
def test_list_servers_undetailed(self):
sl = cs.servers.list(detailed=False)
cs.assert_called('GET', '/servers')
[self.assertTrue(isinstance(s, servers.Server)) for s in sl]
def test_get_server_details(self):
s = cs.servers.get(1234)
cs.assert_called('GET', '/servers/1234')
self.assertTrue(isinstance(s, servers.Server))
self.assertEqual(s.id, 1234)
self.assertEqual(s.status, 'BUILD')
def test_create_server(self):
s = cs.servers.create(
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
ipgroup=1,
files={
'/etc/passwd': 'some data', # a file
'/tmp/foo.txt': StringIO.StringIO('data') # a stream
}
)
cs.assert_called('POST', '/servers')
self.assertTrue(isinstance(s, servers.Server))
def test_update_server(self):
s = cs.servers.get(1234)
# Update via instance
s.update(name='hi')
cs.assert_called('PUT', '/servers/1234')
s.update(name='hi', password='there')
cs.assert_called('PUT', '/servers/1234')
# Silly, but not an error
s.update()
# Update via manager
cs.servers.update(s, name='hi')
cs.assert_called('PUT', '/servers/1234')
cs.servers.update(1234, password='there')
cs.assert_called('PUT', '/servers/1234')
cs.servers.update(s, name='hi', password='there')
cs.assert_called('PUT', '/servers/1234')
def test_delete_server(self):
s = cs.servers.get(1234)
s.delete()
cs.assert_called('DELETE', '/servers/1234')
cs.servers.delete(1234)
cs.assert_called('DELETE', '/servers/1234')
cs.servers.delete(s)
cs.assert_called('DELETE', '/servers/1234')
def test_find(self):
s = cs.servers.find(name='sample-server')
cs.assert_called('GET', '/servers/detail')
self.assertEqual(s.name, 'sample-server')
# Find with multiple results arbitraility returns the first item
s = cs.servers.find(flavorId=1)
sl = cs.servers.findall(flavorId=1)
self.assertEqual(sl[0], s)
self.assertEqual([s.id for s in sl], [1234, 5678])
def test_share_ip(self):
s = cs.servers.get(1234)
# Share via instance
s.share_ip(ipgroup=1, address='1.2.3.4')
cs.assert_called('PUT', '/servers/1234/ips/public/1.2.3.4')
# Share via manager
cs.servers.share_ip(s, ipgroup=1, address='1.2.3.4', configure=False)
cs.assert_called('PUT', '/servers/1234/ips/public/1.2.3.4')
def test_unshare_ip(self):
s = cs.servers.get(1234)
# Unshare via instance
s.unshare_ip('1.2.3.4')
cs.assert_called('DELETE', '/servers/1234/ips/public/1.2.3.4')
# Unshare via manager
cs.servers.unshare_ip(s, '1.2.3.4')
cs.assert_called('DELETE', '/servers/1234/ips/public/1.2.3.4')
def test_reboot_server(self):
s = cs.servers.get(1234)
s.reboot()
cs.assert_called('POST', '/servers/1234/action')
cs.servers.reboot(s, type='HARD')
cs.assert_called('POST', '/servers/1234/action')
def test_rebuild_server(self):
s = cs.servers.get(1234)
s.rebuild(image=1)
cs.assert_called('POST', '/servers/1234/action')
cs.servers.rebuild(s, image=1)
cs.assert_called('POST', '/servers/1234/action')
def test_resize_server(self):
s = cs.servers.get(1234)
s.resize(flavor=1)
cs.assert_called('POST', '/servers/1234/action')
cs.servers.resize(s, flavor=1)
cs.assert_called('POST', '/servers/1234/action')
def test_confirm_resized_server(self):
s = cs.servers.get(1234)
s.confirm_resize()
cs.assert_called('POST', '/servers/1234/action')
cs.servers.confirm_resize(s)
cs.assert_called('POST', '/servers/1234/action')
def test_revert_resized_server(self):
s = cs.servers.get(1234)
s.revert_resize()
cs.assert_called('POST', '/servers/1234/action')
cs.servers.revert_resize(s)
cs.assert_called('POST', '/servers/1234/action')
def test_backup_server(self):
s = cs.servers.get(1234)
s.backup("ImageName", "daily", 10)
cs.assert_called('POST', '/servers/1234/action')
cs.servers.backup(s, "ImageName", "daily", 10)
cs.assert_called('POST', '/servers/1234/action')
def test_migrate_server(self):
s = cs.servers.get(1234)
s.migrate()
cs.assert_called('POST', '/servers/1234/migrate')
cs.servers.migrate(s)
cs.assert_called('POST', '/servers/1234/migrate')
def test_add_fixed_ip(self):
s = cs.servers.get(1234)
s.add_fixed_ip(1)
cs.assert_called('POST', '/servers/1234/action')
cs.servers.add_fixed_ip(s, 1)
cs.assert_called('POST', '/servers/1234/action')
def test_remove_fixed_ip(self):
s = cs.servers.get(1234)
s.remove_fixed_ip('10.0.0.1')
cs.assert_called('POST', '/servers/1234/action')
cs.servers.remove_fixed_ip(s, '10.0.0.1')
cs.assert_called('POST', '/servers/1234/action')
def test_rescue(self):
s = cs.servers.get(1234)
s.rescue()
cs.assert_called('POST', '/servers/1234/rescue')
cs.servers.rescue(s)
cs.assert_called('POST', '/servers/1234/rescue')
def test_unrescue(self):
s = cs.servers.get(1234)
s.unrescue()
cs.assert_called('POST', '/servers/1234/unrescue')
cs.servers.unrescue(s)
cs.assert_called('POST', '/servers/1234/unrescue')

343
tests/v1_0/test_shell.py
View File

@ -1,343 +0,0 @@
import os
import mock
from novaclient import exceptions
import novaclient.shell
from tests import utils
from tests.v1_0 import fakes
class ShellTest(utils.TestCase):
def setUp(self):
"""Run before each test."""
self.old_environment = os.environ.copy()
os.environ = {
'NOVA_USERNAME': 'username',
'NOVA_PASSWORD': 'password',
'NOVA_PROJECT_ID': 'project_id',
'NOVA_VERSION': '1.0',
}
self.shell = novaclient.shell.OpenStackComputeShell()
self.shell.get_api_class = lambda *_: fakes.FakeClient
def tearDown(self):
os.environ = self.old_environment
def run_command(self, cmd):
self.shell.main(cmd.split())
def assert_called(self, method, url, body=None):
return self.shell.cs.assert_called(method, url, body)
def assert_called_anytime(self, method, url, body=None):
return self.shell.cs.assert_called_anytime(method, url, body)
def test_backup_schedule(self):
self.run_command('backup-schedule 1234')
self.assert_called('GET', '/servers/1234/backup_schedule')
self.run_command('backup-schedule sample-server --weekly monday')
self.assert_called(
'POST', '/servers/1234/backup_schedule',
{'backupSchedule': {'enabled': True, 'daily': 'DISABLED',
'weekly': 'MONDAY'}}
)
self.run_command('backup-schedule sample-server '
'--weekly disabled --daily h_0000_0200')
self.assert_called(
'POST', '/servers/1234/backup_schedule',
{'backupSchedule': {'enabled': True, 'daily': 'H_0000_0200',
'weekly': 'DISABLED'}}
)
self.run_command('backup-schedule sample-server --disable')
self.assert_called(
'POST', '/servers/1234/backup_schedule',
{'backupSchedule': {'enabled': False, 'daily': 'DISABLED',
'weekly': 'DISABLED'}}
)
def test_backup_schedule_delete(self):
self.run_command('backup-schedule-delete 1234')
self.assert_called('DELETE', '/servers/1234/backup_schedule')
def test_boot(self):
self.run_command('boot --image 1 some-server')
self.assert_called(
'POST', '/servers',
{'server': {'flavorId': 1, 'name': 'some-server', 'imageId': 1,
'min_count': 1, 'max_count': 1}}
)
self.run_command('boot --image 1 --meta foo=bar'
' --meta spam=eggs some-server ')
self.assert_called(
'POST', '/servers',
{'server': {'flavorId': 1, 'name': 'some-server', 'imageId': 1,
'min_count': 1, 'max_count': 1,
'metadata': {'foo': 'bar', 'spam': 'eggs'}}}
)
def test_boot_files(self):
testfile = os.path.join(os.path.dirname(__file__), 'testfile.txt')
expected_file_data = open(testfile).read().encode('base64')
self.run_command('boot some-server --image 1 '
'--file /tmp/foo=%s --file /tmp/bar=%s' %
(testfile, testfile))
self.assert_called(
'POST', '/servers',
{'server': {'flavorId': 1, 'name': 'some-server', 'imageId': 1,
'min_count': 1, 'max_count': 1,
'personality': [
{'path': '/tmp/bar', 'contents': expected_file_data},
{'path': '/tmp/foo', 'contents': expected_file_data}
]}
}
)
def test_boot_invalid_file(self):
invalid_file = os.path.join(os.path.dirname(__file__),
'asdfasdfasdfasdf')
self.assertRaises(exceptions.CommandError, self.run_command,
'boot some-server --image 1 '
'--file /foo=%s' % invalid_file)
def test_boot_key_auto(self):
mock_exists = mock.Mock(return_value=True)
mock_open = mock.Mock()
mock_open.return_value = mock.Mock()
mock_open.return_value.read = mock.Mock(return_value='SSHKEY')
@mock.patch('os.path.exists', mock_exists)
@mock.patch('__builtin__.open', mock_open)
def test_shell_call():
self.run_command('boot some-server --image 1 --key')
self.assert_called(
'POST', '/servers',
{'server': {'flavorId': 1, 'name': 'some-server',
'imageId': 1, 'min_count': 1, 'max_count': 1,
'personality': [{
'path': '/root/.ssh/authorized_keys2',
'contents': ('SSHKEY').encode('base64')},
]}
}
)
test_shell_call()
def test_boot_key_auto_no_keys(self):
mock_exists = mock.Mock(return_value=False)
@mock.patch('os.path.exists', mock_exists)
def test_shell_call():
self.assertRaises(exceptions.CommandError, self.run_command,
'boot some-server --image 1 --key')
test_shell_call()
def test_boot_key_file(self):
testfile = os.path.join(os.path.dirname(__file__), 'testfile.txt')
expected_file_data = open(testfile).read().encode('base64')
self.run_command('boot some-server --image 1 --key %s' % testfile)
self.assert_called(
'POST', '/servers',
{'server': {'flavorId': 1, 'name': 'some-server', 'imageId': 1,
'min_count': 1, 'max_count': 1,
'personality': [
{'path': '/root/.ssh/authorized_keys2', 'contents':
expected_file_data},
]}
}
)
def test_boot_invalid_keyfile(self):
invalid_file = os.path.join(os.path.dirname(__file__),
'asdfasdfasdfasdf')
self.assertRaises(exceptions.CommandError, self.run_command,
'boot some-server --image 1 --key %s' % invalid_file)
def test_boot_ipgroup(self):
self.run_command('boot --image 1 --ipgroup 1 some-server')
self.assert_called(
'POST', '/servers',
{'server': {'flavorId': 1, 'name': 'some-server', 'imageId': 1,
'sharedIpGroupId': 1, 'min_count': 1, 'max_count': 1}}
)
def test_boot_ipgroup_name(self):
self.run_command('boot --image 1 --ipgroup group1 some-server')
self.assert_called(
'POST', '/servers',
{'server': {'flavorId': 1, 'name': 'some-server', 'imageId': 1,
'sharedIpGroupId': 1, 'min_count': 1, 'max_count': 1}}
)
def test_flavor_list(self):
self.run_command('flavor-list')
self.assert_called_anytime('GET', '/flavors/detail')
def test_image_list(self):
self.run_command('image-list')
self.assert_called('GET', '/images/detail')
def test_snapshot_create(self):
self.run_command('image-create sample-server mysnapshot')
self.assert_called(
'POST', '/images',
{'image': {'name': 'mysnapshot', 'serverId': 1234}}
)
def test_image_delete(self):
self.run_command('image-delete 1')
self.assert_called('DELETE', '/images/1')
def test_ip_share(self):
self.run_command('ip-share sample-server 1 1.2.3.4')
self.assert_called(
'PUT', '/servers/1234/ips/public/1.2.3.4',
{'shareIp': {'sharedIpGroupId': 1, 'configureServer': True}}
)
def test_ip_unshare(self):
self.run_command('ip-unshare sample-server 1.2.3.4')
self.assert_called('DELETE', '/servers/1234/ips/public/1.2.3.4')
def test_ipgroup_list(self):
self.run_command('ipgroup-list')
assert ('GET', '/shared_ip_groups/detail', None) in \
self.shell.cs.client.callstack
self.assert_called('GET', '/servers/5678')
def test_ipgroup_show(self):
self.run_command('ipgroup-show 1')
self.assert_called('GET', '/shared_ip_groups/1')
self.run_command('ipgroup-show group2')
# does a search, not a direct GET
self.assert_called('GET', '/shared_ip_groups/detail')
def test_ipgroup_create(self):
self.run_command('ipgroup-create a-group')
self.assert_called(
'POST', '/shared_ip_groups',
{'sharedIpGroup': {'name': 'a-group'}}
)
self.run_command('ipgroup-create a-group sample-server')
self.assert_called(
'POST', '/shared_ip_groups',
{'sharedIpGroup': {'name': 'a-group', 'server': 1234}}
)
def test_ipgroup_delete(self):
self.run_command('ipgroup-delete group1')
self.assert_called('DELETE', '/shared_ip_groups/1')
def test_list(self):
self.run_command('list')
self.assert_called('GET', '/servers/detail')
def test_reboot(self):
self.run_command('reboot sample-server')
self.assert_called('POST', '/servers/1234/action',
{'reboot': {'type': 'SOFT'}})
self.run_command('reboot sample-server --hard')
self.assert_called('POST', '/servers/1234/action',
{'reboot': {'type': 'HARD'}})
def test_rebuild(self):
self.run_command('rebuild sample-server 1')
self.assert_called('POST', '/servers/1234/action',
{'rebuild': {'imageId': 1}})
def test_rename(self):
self.run_command('rename sample-server newname')
self.assert_called('PUT', '/servers/1234',
{'server': {'name': 'newname'}})
def test_resize(self):
self.run_command('resize sample-server 1')
self.assert_called('POST', '/servers/1234/action',
{'resize': {'flavorId': 1}})
def test_resize_confirm(self):
self.run_command('resize-confirm sample-server')
self.assert_called('POST', '/servers/1234/action',
{'confirmResize': None})
def test_resize_revert(self):
self.run_command('resize-revert sample-server')
self.assert_called('POST', '/servers/1234/action',
{'revertResize': None})
def test_backup(self):
self.run_command('backup sample-server mybackup daily 1')
self.assert_called(
'POST', '/servers/1234/action',
{'createBackup': {'name': 'mybackup', 'backup_type': 'daily',
'rotation': 1}}
)
@mock.patch('getpass.getpass', mock.Mock(return_value='p'))
def test_root_password(self):
self.run_command('root-password sample-server')
self.assert_called('PUT', '/servers/1234',
{'server': {'adminPass': 'p'}})
def test_show(self):
self.run_command('show 1234')
# XXX need a way to test multiple calls
# self.assert_called('GET', '/servers/1234')
self.assert_called('GET', '/images/2')
def test_delete(self):
self.run_command('delete 1234')
self.assert_called('DELETE', '/servers/1234')
self.run_command('delete sample-server')
self.assert_called('DELETE', '/servers/1234')
def test_zone(self):
self.run_command('zone 1')
self.assert_called('GET', '/zones/1')
self.run_command('zone 1 --api_url=http://zzz '
'--zone_username=frank --zone_password=xxx')
self.assert_called(
'PUT', '/zones/1',
{'zone': {'username': 'frank', 'password': 'xxx',
'api_url': 'http://zzz'}}
)
def test_zone_add(self):
self.run_command('zone-add child_zone http://zzz '
'--zone_username=frank --zone_password=xxx '
'--weight_offset=0.0 --weight_scale=1.0')
self.assert_called(
'POST', '/zones',
{'zone': {'name': 'child_zone',
'api_url': 'http://zzz', 'username': 'frank',
'password': 'xxx',
'weight_offset': '0.0', 'weight_scale': '1.0'}}
)
def test_zone_add_optional(self):
self.run_command('zone-add child_zone http://zzz')
self.assert_called(
'POST', '/zones',
{'zone': {'name': 'child_zone',
'api_url': 'http://zzz',
'username': None,
'password': None,
'weight_offset': 0.0, 'weight_scale': 1.0}}
)
def test_zone_delete(self):
self.run_command('zone-delete 1')
self.assert_called('DELETE', '/zones/1')
def test_zone_list(self):
self.run_command('zone-list')
assert ('GET', '/zones/detail', None) in self.shell.cs.client.callstack

76
tests/v1_0/test_zones.py
View File

@ -1,76 +0,0 @@
from novaclient.v1_0 import zones
from tests import utils
from tests.v1_0 import fakes
os = fakes.FakeClient()
class ZonesTest(utils.TestCase):
def test_list_zones(self):
sl = os.zones.list()
os.assert_called('GET', '/zones/detail')
[self.assertTrue(isinstance(s, zones.Zone)) for s in sl]
def test_list_zones_undetailed(self):
sl = os.zones.list(detailed=False)
os.assert_called('GET', '/zones')
[self.assertTrue(isinstance(s, zones.Zone)) for s in sl]
def test_get_zone_details(self):
s = os.zones.get(1)
os.assert_called('GET', '/zones/1')
self.assertTrue(isinstance(s, zones.Zone))
self.assertEqual(s.id, 1)
self.assertEqual(s.api_url, 'http://foo.com')
def test_create_zone(self):
s = os.zones.create(zone_name='child_zone',
api_url="http://foo.com",
username='bob',
password='xxx')
os.assert_called('POST', '/zones')
self.assertTrue(isinstance(s, zones.Zone))
def test_update_zone(self):
s = os.zones.get(1)
# Update via instance
s.update(api_url='http://blah.com')
os.assert_called('PUT', '/zones/1')
s.update(api_url='http://blah.com', username='alice', password='xxx')
os.assert_called('PUT', '/zones/1')
# Silly, but not an error
s.update()
# Update via manager
os.zones.update(s, api_url='http://blah.com')
os.assert_called('PUT', '/zones/1')
os.zones.update(1, api_url='http://blah.com')
os.assert_called('PUT', '/zones/1')
os.zones.update(s, api_url='http://blah.com', username='fred',
password='zip')
os.assert_called('PUT', '/zones/1')
def test_delete_zone(self):
s = os.zones.get(1)
s.delete()
os.assert_called('DELETE', '/zones/1')
os.zones.delete(1)
os.assert_called('DELETE', '/zones/1')
os.zones.delete(s)
os.assert_called('DELETE', '/zones/1')
def test_find_zone(self):
s = os.zones.find(password='qwerty')
os.assert_called('GET', '/zones/detail')
self.assertEqual(s.username, 'bob')
# Find with multiple results returns the first item
s = os.zones.find(api_url='http://foo.com')
sl = os.zones.findall(api_url='http://foo.com')
self.assertEqual(sl[0], s)
self.assertEqual([s.id for s in sl], [1, 2])

1
tests/v1_0/testfile.txt
View File

@ -1 +0,0 @@
BLAH

28
tests/v1_0/utils.py
View File

@ -1,28 +0,0 @@
from nose.tools import ok_
def fail(msg):
raise AssertionError(msg)
def assert_in(thing, seq, msg=None):
msg = msg or "'%s' not found in %s" % (thing, seq)
ok_(thing in seq, msg)
def assert_not_in(thing, seq, msg=None):
msg = msg or "unexpected '%s' found in %s" % (thing, seq)
ok_(thing not in seq, msg)
def assert_has_keys(dict, required=[], optional=[]):
keys = dict.keys()
for k in required:
assert_in(k, keys, "required key %s missing from %s" % (k, dict))
extra_keys = set(keys).difference(set(required + optional))
if extra_keys:
fail("found unexpected keys: %s" % list(extra_keys))
def assert_isinstance(thing, kls):
ok_(isinstance(thing, kls), "%s is not an instance of %s" % (thing, kls))