openstack
/
openstacksdk
Code Issues Proposed changes

Refactor compute for new resource/proxy 

This change refactors the compute resources and proxy to work with the
newly refactored classes.

Additionally, it adds several properties to the Server resource that
have been documented since it was originally implemented. Several of
these have been requested lately by users, including security_groups,
availability_zone, and admin_password.

Change-Id: I387c9573234c3ed08eb966e388cc2e1a3d891c2d
This commit is contained in:
Brian Curtin 2016-03-23 22:54:13 -04:00 committed by Brian Curtin
parent 586d9afdab
commit ccd8daa632
29 changed files with 907 additions and 691 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

323
openstack/compute/v2/_proxy.py
View File

@ -21,11 +21,11 @@ from openstack.compute.v2 import server as _server
from openstack.compute.v2 import server_group as _server_group
from openstack.compute.v2 import server_interface as _server_interface
from openstack.compute.v2 import server_ip
from openstack import proxy
from openstack import resource
from openstack import proxy2
from openstack import resource2
class Proxy(proxy.BaseProxy):
class Proxy(proxy2.BaseProxy):
def find_extension(self, name_or_id, ignore_missing=True):
"""Find a single extension
@ -42,16 +42,13 @@ class Proxy(proxy.BaseProxy):
return self._find(extension.Extension, name_or_id,
ignore_missing=ignore_missing)
def extensions(self, **query):
def extensions(self):
"""Retrieve a generator of extensions
:param kwargs \*\*query: Optional query parameters to be sent to limit
the resources being returned.
:returns: A generator of extension instances.
:rtype: :class:`~openstack.compute.v2.extension.Extension`
"""
return self._list(extension.Extension, paginated=False, **query)
return self._list(extension.Extension, paginated=False)
def find_flavor(self, name_or_id, ignore_missing=True):
"""Find a single flavor
@ -197,7 +194,7 @@ class Proxy(proxy.BaseProxy):
if isinstance(res, base):
return res
else:
return base({"id": res})
return base(id=res)
def get_image_metadata(self, image):
"""Return a dictionary of metadata for an image
@ -308,30 +305,13 @@ class Proxy(proxy.BaseProxy):
return self._find(_keypair.Keypair, name_or_id,
ignore_missing=ignore_missing)
def keypairs(self, **query):
def keypairs(self):
"""Return a generator of keypairs
:param kwargs \*\*query: Optional query parameters to be sent to limit
the resources being returned.
:returns: A generator of keypair objects
:rtype: :class:`~openstack.compute.v2.keypair.Keypair`
"""
return self._list(_keypair.Keypair, paginated=False, **query)
def update_keypair(self, keypair, **attrs):
"""Update a keypair
:param keypair: Either the ID of a keypair or a
:class:`~openstack.compute.v2.keypair.Keypair`
instance.
:attrs kwargs: The attributes to update on the keypair represented
by ``keypair``.
:returns: The updated keypair
:rtype: :class:`~openstack.compute.v2.keypair.Keypair`
"""
return self._update(_keypair.Keypair, keypair, **attrs)
return self._list(_keypair.Keypair, paginated=False)
def get_limits(self):
"""Retrieve limits that are applied to the project's account
@ -434,12 +414,6 @@ class Proxy(proxy.BaseProxy):
:returns: A generator of server instances.
"""
srv = _server.ServerDetail if details else _server.Server
# Server expects changes-since, but we use an underscore
# so it can be a proper Python name.
if "changes_since" in query:
query["changes-since"] = query.pop("changes_since")
return self._list(srv, paginated=True, **query)
def update_server(self, server, **attrs):
@ -455,10 +429,142 @@ class Proxy(proxy.BaseProxy):
"""
return self._update(_server.Server, server, **attrs)
def change_server_password(self, server, new_password):
"""Change the administrator password
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param str new_password: The new password to be set.
:returns: None
"""
server = self._get_resource(_server.Server, server)
server.change_password(self.session, new_password)
def reboot_server(self, server, reboot_type):
"""Reboot a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param str reboot_type: The type of reboot to perform.
"HARD" and "SOFT" are the current options.
:returns: None
"""
server = self._get_resource(_server.Server, server)
server.reboot(self.session, reboot_type)
def rebuild_server(self, server, name, admin_password, **attrs):
"""Rebuild a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param str name: The name of the server
:param str admin_password: The administrator password
:param bool preserve_ephemeral: Indicates whether the server
is rebuilt with the preservation of the ephemeral partition.
*Default: False*
:param str image: The id of an image to rebuild with. *Default: None*
:param str access_ipv4: The IPv4 address to rebuild with.
*Default: None*
:param str access_ipv6: The IPv6 address to rebuild with.
*Default: None*
:param dict metadata: A dictionary of metadata to rebuild with.
*Default: None*
:param list personality: A list of dictionaries, each including a
**path** and **contents** key, to be injected
into the rebuilt server at launch.
*Default: None*
:returns: The rebuilt :class:`~openstack.compute.v2.server.Server`
instance.
"""
server = self._get_resource(_server.Server, server)
return server.rebuild(self.session, name, admin_password, **attrs)
def resize_server(self, server, flavor):
"""Resize a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param flavor: Either the ID of a flavor or a
:class:`~openstack.compute.v2.flavor.Flavor` instance.
:returns: None
"""
server = self._get_resource(_server.Server, server)
flavor_id = resource2.Resource._get_id(flavor)
server.resize(self.session, flavor_id)
def confirm_server_resize(self, server):
"""Confirm a server resize
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:returns: None
"""
server = self._get_resource(_server.Server, server)
server.confirm_resize(self.session)
def revert_server_resize(self, server):
"""Revert a server resize
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:returns: None
"""
server = self._get_resource(_server.Server, server)
server.revert_resize(self.session)
def create_server_image(self, server, name, metadata=None):
"""Create an image from a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param str name: The name of the image to be created.
:param dict metadata: A dictionary of metadata to be set on the image.
:returns: None
"""
server = self._get_resource(_server.Server, server)
server.create_image(self.session, name, metadata)
def add_security_group_to_server(self, server, security_group):
"""Add a security group to a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param security_group: Either the ID of a security group or a
:class:`~openstack.network.v2.security_group.SecurityGroup`
instance.
:returns: None
"""
server = self._get_resource(_server.Server, server)
security_group_id = resource2.Resource._get_id(security_group)
server.add_security_group(self.session, security_group_id)
def remove_security_group_from_server(self, server, security_group):
"""Add a security group to a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param security_group: Either the ID of a security group or a
:class:`~openstack.network.v2.security_group.SecurityGroup`
instance.
:returns: None
"""
server = self._get_resource(_server.Server, server)
security_group_id = resource2.Resource._get_id(security_group)
server.remove_security_group(self.session, security_group_id)
def wait_for_server(self, server, status='ACTIVE', failures=['ERROR'],
interval=2, wait=120):
return resource.wait_for_status(self.session, server, status,
failures, interval, wait)
return resource2.wait_for_status(self.session, server, status,
failures, interval, wait)
def create_server_interface(self, server, **attrs):
"""Create a new server interface from attributes
@ -473,9 +579,9 @@ class Proxy(proxy.BaseProxy):
:returns: The results of server interface creation
:rtype: :class:`~openstack.compute.v2.server_interface.ServerInterface`
"""
server_id = resource.Resource.get_id(server)
server_id = resource2.Resource._get_id(server)
return self._create(_server_interface.ServerInterface,
path_args={'server_id': server_id}, **attrs)
server_id=server_id, **attrs)
def delete_server_interface(self, server_interface, server=None,
ignore_missing=True):
@ -497,13 +603,13 @@ class Proxy(proxy.BaseProxy):
:returns: ``None``
"""
if isinstance(server_interface, _server_interface.ServerInterface):
server_id = server_interface.server_id
else:
server_id = resource.Resource.get_id(server)
server_id = self._get_uri_attribute(server_interface, server,
"server_id")
server_interface = resource2.Resource._get_id(server_interface)
self._delete(_server_interface.ServerInterface, server_interface,
path_args={'server_id': server_id},
self._delete(_server_interface.ServerInterface,
port_id=server_interface,
server_id=server_id,
ignore_missing=ignore_missing)
def get_server_interface(self, server_interface, server=None):
@ -523,131 +629,58 @@ class Proxy(proxy.BaseProxy):
:raises: :class:`~openstack.exceptions.ResourceNotFound`
when no resource can be found.
"""
if isinstance(server_interface, _server_interface.ServerInterface):
server_id = server_interface.server_id
else:
server_id = resource.Resource.get_id(server)
server_id = self._get_uri_attribute(server_interface, server,
"server_id")
server_interface = resource2.Resource._get_id(server_interface)
return self._get(_server_interface.ServerInterface, server_interface,
path_args={'server_id': server_id})
return self._get(_server_interface.ServerInterface,
server_id=server_id, port_id=server_interface)
def server_interfaces(self, server, **query):
def server_interfaces(self, server):
"""Return a generator of server interfaces
:param server: The server can be either the ID of a server or a
:class:`~openstack.compute.v2.server.Server`.
:param kwargs \*\*query: Optional query parameters to be sent to limit
the resources being returned.
:returns: A generator of ServerInterface objects
:rtype: :class:`~openstack.compute.v2.server_interface.ServerInterface`
"""
server_id = resource.Resource.get_id(server)
server_id = resource2.Resource._get_id(server)
return self._list(_server_interface.ServerInterface, paginated=False,
path_args={'server_id': server_id},
**query)
server_id=server_id)
def find_server_ip(self, name_or_id, ignore_missing=True):
"""Find a single server IP
:param name_or_id: The name or ID of a server IP.
:param bool ignore_missing: When set to ``False``
:class:`~openstack.exceptions.ResourceNotFound` will be
raised when the resource does not exist.
When set to ``True``, None will be returned when
attempting to find a nonexistent resource.
:returns: One :class:`~openstack.compute.v2.server_ip.ServerIP` or None
"""
return self._find(server_ip.ServerIP, name_or_id,
ignore_missing=ignore_missing)
def server_ips(self, **query):
def server_ips(self, server, network_label=None):
"""Return a generator of server IPs
:param kwargs \*\*query: Optional query parameters to be sent to limit
the resources being returned.
:param server: The server can be either the ID of a server or a
:class:`~openstack.compute.v2.server.Server`.
:param network_label: The name of a particular network to list
IP addresses from.
:returns: A generator of ServerIP objects
:rtype: :class:`~openstack.compute.v2.server_ip.ServerIP`
"""
return self._list(server_ip.ServerIP, paginated=False, **query)
server_id = resource2.Resource._get_id(server)
return self._list(server_ip.ServerIP, paginated=False,
server_id=server_id, network_label=network_label)
def resize_server(self, server, flavor):
"""Resize a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param falvor: The ID or name of the flavor used to resize the server.
:returns: None
"""
server = _server.Server.from_id(server)
server.resize(self.session, flavor)
def confirm_resize_server(self, server):
"""Confirm a pending resize_server action
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:returns: None
"""
server = _server.Server.from_id(server)
server.confirm_resize(self.session)
def revert_resize_server(self, server):
"""Cancel and revert a pending resize_server action
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:returns: None
"""
server = _server.Server.from_id(server)
server.revert_resize(self.session)
def rebuild_server(self, server, image, name=None, admin_password=None,
**attrs):
"""Rebuild a server
:param server: Either the ID of a server or a
:class:`~openstack.compute.v2.server.Server` instance.
:param image: The ID or name or a
:class:`~openstack.compute.v2.image.Image` or full
URL of the image used to rebuild the server with.
:param name: New name for the server.
:param admin_password: New admin password for the server.
:param kwargs \*\*attrs: The attributes to rebuild the server.
:returns: The rebuilt server
:rtype: :class:`~openstack.compute.v2.server.Server`
"""
if isinstance(image, _image.Image):
image_ref = image.id
else:
image_obj = self.find_image(image)
if image_obj:
image_ref = image_obj.id
else:
# the 'image' could be a full url
image_ref = image
server = _server.Server.from_id(server)
return server.rebuild(self.session, name, image_ref, admin_password,
**attrs)
def availability_zones(self, **query):
def availability_zones(self, details=False):
"""Return a generator of availability zones
:param kwargs \*\*query: Optional query parameters to be sent
to limit the resources being returned.
:param bool details: Return extra details about the availability
zones. This defaults to `False` as it generally
requires extra permission.
:returns: A generator of availability zone
:rtype: :class:`~openstack.compute.v2.availability_zone.
AvailabilityZone`
"""
return self._list(availability_zone.AvailabilityZone,
paginated=False, **query)
if details:
az = availability_zone.AvailabilityZoneDetail
else:
az = availability_zone.AvailabilityZone
return self._list(az, paginated=False)
def get_server_metadata(self, server):
"""Return a dictionary of metadata for a server
@ -773,14 +806,14 @@ class Proxy(proxy.BaseProxy):
"""
return self._list(_server_group.ServerGroup, paginated=False, **query)
def hypervisors(self, **query):
def hypervisors(self):
"""Return a generator of hypervisor
:returns: A generator of hypervisor
:rtype: class: `~openstack.compute.v2.hypervisor.Hypervisor`
"""
return self._list(_hypervisor.Hypervisor, paginated=False, **query)
return self._list(_hypervisor.Hypervisor, paginated=False)
def find_hypervisor(self, name_or_id, ignore_missing=True):
"""Find a hypervisor from name or id to get the corresponding info

18
openstack/compute/v2/availability_zone.py
View File

@ -11,12 +11,10 @@
# under the License.
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class AvailabilityZone(resource.Resource):
resource_key = 'availability_zone'
class AvailabilityZone(resource2.Resource):
resources_key = 'availabilityZoneInfo'
base_path = '/os-availability-zone'
@ -27,10 +25,12 @@ class AvailabilityZone(resource.Resource):
# Properties
#: name of availability zone
name = resource.prop('zoneName')
name = resource2.Body('zoneName')
#: state of availability zone
state = resource.prop('zoneState')
state = resource2.Body('zoneState')
#: hosts of availability zone
hosts = resource.prop('hosts')
hosts = resource2.Body('hosts')
class AvailabilityZoneDetail(AvailabilityZone):
base_path = '/os-availability-zone/detail'

18
openstack/compute/v2/extension.py
View File

@ -11,10 +11,10 @@
# under the License.
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class Extension(resource.Resource):
class Extension(resource2.Resource):
resource_key = 'extension'
resources_key = 'extensions'
base_path = '/extensions'
@ -22,20 +22,20 @@ class Extension(resource.Resource):
id_attribute = "alias"
# capabilities
allow_retrieve = True
allow_get = True
allow_list = True
# Properties
#: A short name by which this extension is also known.
alias = resource.prop('alias')
alias = resource2.Body('alias', alternate_id=True)
#: Text describing this extension's purpose.
description = resource.prop('description')
description = resource2.Body('description')
#: Links pertaining to this extension. This is a list of dictionaries,
#: each including keys ``href`` and ``rel``.
links = resource.prop('links')
links = resource2.Body('links')
#: The name of the extension.
name = resource.prop('name')
name = resource2.Body('name')
#: A URL pointing to the namespace for this extension.
namespace = resource.prop('namespace')
namespace = resource2.Body('namespace')
#: Timestamp when this extension was last updated.
updated_at = resource.prop('updated')
updated_at = resource2.Body('updated')

32
openstack/compute/v2/flavor.py
View File

@ -11,10 +11,10 @@
# under the License.
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class Flavor(resource.Resource):
class Flavor(resource2.Resource):
resource_key = 'flavor'
resources_key = 'flavors'
base_path = '/flavors'
@ -22,41 +22,45 @@ class Flavor(resource.Resource):
# capabilities
allow_create = True
allow_retrieve = True
allow_get = True
allow_update = True
allow_delete = True
allow_list = True
_query_mapping = resource2.QueryParameters("sort_key", "sort_dir",
min_disk="minDisk",
min_ram="minRam")
# Properties
#: Links pertaining to this flavor. This is a list of dictionaries,
#: each including keys ``href`` and ``rel``.
links = resource.prop('links')
links = resource2.Body('links')
#: The name of this flavor.
name = resource.prop('name')
name = resource2.Body('name')
#: Size of the disk this flavor offers. *Type: int*
disk = resource.prop('disk', type=int)
disk = resource2.Body('disk', type=int)
#: ``True`` if this is a publicly visible flavor. ``False`` if this is
#: a private image. *Type: bool*
is_public = resource.prop('os-flavor-access:is_public', type=bool)
is_public = resource2.Body('os-flavor-access:is_public', type=bool)
#: The amount of RAM (in MB) this flavor offers. *Type: int*
ram = resource.prop('ram', type=int)
ram = resource2.Body('ram', type=int)
#: The number of virtual CPUs this flavor offers. *Type: int*
vcpus = resource.prop('vcpus', type=int)
vcpus = resource2.Body('vcpus', type=int)
#: Size of the swap partitions.
swap = resource.prop('swap')
swap = resource2.Body('swap')
#: Size of the ephemeral data disk attached to this server. *Type: int*
ephemeral = resource.prop('OS-FLV-EXT-DATA:ephemeral', type=int)
ephemeral = resource2.Body('OS-FLV-EXT-DATA:ephemeral', type=int)
#: ``True`` if this flavor is disabled, ``False`` if not. *Type: bool*
is_disabled = resource.prop('OS-FLV-DISABLED:disabled', type=bool)
is_disabled = resource2.Body('OS-FLV-DISABLED:disabled', type=bool)
#: The bandwidth scaling factor this flavor receives on the network.
rxtx_factor = resource.prop('rxtx_factor', type=float)
rxtx_factor = resource2.Body('rxtx_factor', type=float)
class FlavorDetail(Flavor):
base_path = '/flavors/detail'
allow_create = False
allow_retrieve = False
allow_get = False
allow_update = False
allow_delete = False
allow_list = True

54
openstack/compute/v2/hypervisor.py
View File

@ -12,12 +12,10 @@
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class Hypervisor(resource.Resource):
name_attribute = 'hypervisor_hostname'
class Hypervisor(resource2.Resource):
resource_key = 'hypervisor'
resources_key = 'hypervisors'
base_path = '/os-hypervisors'
@ -25,15 +23,45 @@ class Hypervisor(resource.Resource):
service = compute_service.ComputeService()
# capabilities
allow_retrieve = True
allow_get = True
allow_list = True
# Properties
#: status of hypervisor
status = resource.prop('status')
#: state of hypervisor
state = resource.prop('state')
#: name of hypervisor
hypervisor_hostname = resource.prop('hypervisor_hostname')
#: Status of hypervisor
status = resource2.Body('status')
#: State of hypervisor
state = resource2.Body('state')
#: Name of hypervisor
name = resource2.Body('hypervisor_hostname')
#: Service details
service_details = resource2.Body('service')
#: Count of the VCPUs in use
vcpus_used = resource2.Body('vcpus_used')
#: Count of all VCPUs
vcpus = resource2.Body('vcpus')
#: Count of the running virtual machines
running_vms = resource2.Body('running_vms')
#: The type of hypervisor
hypervisor_type = resource2.Body('hypervisor_type')
#: Version of the hypervisor
hypervisor_version = resource2.Body('hypervisor_version')
#: The amount, in gigabytes, of local storage used
local_disk_used = resource2.Body('local_gb_used')
#: The amount, in gigabytes, of the local storage device
local_disk_size = resource2.Body('local_gb')
#: The amount, in gigabytes, of free space on the local storage device
local_disk_free = resource2.Body('free_disk_gb')
#: The amount, in megabytes, of memory
memory_used = resource2.Body('memory_mb_used')
#: The amount, in megabytes, of total memory
memory_size = resource2.Body('memory_mb')
#: The amount, in megabytes, of available memory
memory_free = resource2.Body('free_ram_mb')
#: Measurement of the hypervisor's current workload
current_workload = resource2.Body('current_workload')
#: Information about the hypervisor's CPU
cpu_info = resource2.Body('cpu_info')
#: IP address of the host
host_ip = resource2.Body('host_ip')
#: Disk space available to the scheduler
disk_available = resource2.Body("disk_available_least")

34
openstack/compute/v2/image.py
View File

@ -12,48 +12,54 @@
from openstack.compute import compute_service
from openstack.compute.v2 import metadata
from openstack import resource
from openstack import resource2
class Image(resource.Resource, metadata.MetadataMixin):
class Image(resource2.Resource, metadata.MetadataMixin):
resource_key = 'image'
resources_key = 'images'
base_path = '/images'
service = compute_service.ComputeService()
# capabilities
allow_retrieve = True
allow_get = True
allow_delete = True
allow_list = True
_query_mapping = resource2.QueryParameters("server", "name",
"status", "type",
min_disk="minDisk",
min_ram="minRam",
changes_since="changes-since")
# Properties
#: Links pertaining to this image. This is a list of dictionaries,
#: each including keys ``href`` and ``rel``, and optionally ``type``.
links = resource.prop('links')
links = resource2.Body('links')
#: The name of this image.
name = resource.prop('name')
name = resource2.Body('name')
#: Timestamp when the image was created.
created_at = resource.prop('created')
created_at = resource2.Body('created')
#: Metadata pertaining to this image. *Type: dict*
metadata = resource.prop('metadata', type=dict)
metadata = resource2.Body('metadata', type=dict)
#: The mimimum disk size. *Type: int*
min_disk = resource.prop('minDisk', type=int)
min_disk = resource2.Body('minDisk', type=int)
#: The minimum RAM size. *Type: int*
min_ram = resource.prop('minRam', type=int)
min_ram = resource2.Body('minRam', type=int)
#: If this image is still building, its progress is represented here.
#: Once an image is created, progres will be 100. *Type: int*
progress = resource.prop('progress', type=int)
progress = resource2.Body('progress', type=int)
#: The status of this image.
status = resource.prop('status')
status = resource2.Body('status')
#: Timestamp when the image was updated.
updated_at = resource.prop('updated')
updated_at = resource2.Body('updated')
#: Size of the image in bytes. *Type: int*
size = resource.prop('OS-EXT-IMG-SIZE:size', type=int)
size = resource2.Body('OS-EXT-IMG-SIZE:size', type=int)
class ImageDetail(Image):
base_path = '/images/detail'
allow_retrieve = False
allow_get = False
allow_delete = False
allow_list = True

49
openstack/compute/v2/keypair.py
View File

@ -11,12 +11,10 @@
# under the License.
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class Keypair(resource.Resource):
id_attribute = 'name'
name_attribute = None
class Keypair(resource2.Resource):
resource_key = 'keypair'
resources_key = 'keypairs'
base_path = '/os-keypairs'
@ -24,35 +22,36 @@ class Keypair(resource.Resource):
# capabilities
allow_create = True
allow_retrieve = True
allow_update = True
allow_get = True
allow_delete = True
allow_list = True
# Properties
#: The short fingerprint associated with the ``public_key`` for
#: this keypair.
fingerprint = resource.prop('fingerprint')
fingerprint = resource2.Body('fingerprint')
# NOTE: There is in fact an 'id' field. However, it's not useful
# because all operations use the 'name' as an identifier.
# Additionally, the 'id' field only appears *after* creation,
# so suddenly you have an 'id' field filled in after the fact,
# and it just gets in the way. We need to cover this up by having
# the name be both our id and name.
#: The id identifying the keypair
id = resource2.Body('name')
#: A name identifying the keypair
name = resource.prop('name')
name = resource2.Body('name', alternate_id=True)
#: The private key for the keypair
private_key = resource.prop('private_key')
private_key = resource2.Body('private_key')
#: The SSH public key that is paired with the server.
public_key = resource.prop('public_key')
public_key = resource2.Body('public_key')
def __init__(self, attrs=None, loaded=False):
if attrs is not None:
if 'keypair' in attrs:
attrs = attrs['keypair']
super(Keypair, self).__init__(attrs, loaded=loaded)
@classmethod
def list(cls, session, paginated=False):
resp = session.get(cls.base_path, endpoint_filter=cls.service,
headers={"Accept": "application/json"})
resp = resp.json()
resp = resp[cls.resources_key]
def create(self, session):
"""Create a new keypair from this instance.
This is needed because the name is the id, but we can't create one
with a PUT. That and we need the private_key out of the response.
"""
resp = self.create_by_id(session, self._attrs)
self._attrs = resp
self._reset_dirty()
return self
for data in resp:
value = cls.existing(**data[cls.resource_key])
yield value

91
openstack/compute/v2/limits.py
View File

@ -11,96 +11,99 @@
# under the License.
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class AbsoluteLimits(resource.Resource):
class AbsoluteLimits(resource2.Resource):
#: The number of key-value pairs that can be set as image metadata.
image_meta = resource.prop("maxImageMeta")
image_meta = resource2.Body("maxImageMeta")
#: The maximum number of personality contents that can be supplied.
personality = resource.prop("maxPersonality")
personality = resource2.Body("maxPersonality")
#: The maximum size, in bytes, of a personality.
personality_size = resource.prop("maxPersonalitySize")
personality_size = resource2.Body("maxPersonalitySize")
#: The maximum amount of security group rules allowed.
security_group_rules = resource.prop("maxSecurityGroupRules")
security_group_rules = resource2.Body("maxSecurityGroupRules")
#: The maximum amount of security groups allowed.
security_groups = resource.prop("maxSecurityGroups")
security_groups = resource2.Body("maxSecurityGroups")
#: The amount of security groups currently in use.
security_groups_used = resource.prop("totalSecurityGroupsUsed")
security_groups_used = resource2.Body("totalSecurityGroupsUsed")
#: The number of key-value pairs that can be set as sever metadata.
server_meta = resource.prop("maxServerMeta")
server_meta = resource2.Body("maxServerMeta")
#: The maximum amount of cores.
total_cores = resource.prop("maxTotalCores")
total_cores = resource2.Body("maxTotalCores")
#: The amount of cores currently in use.
total_cores_used = resource.prop("totalCoresUsed")
total_cores_used = resource2.Body("totalCoresUsed")
#: The maximum amount of floating IPs.
floating_ips = resource.prop("maxTotalFloatingIps")
floating_ips = resource2.Body("maxTotalFloatingIps")
#: The amount of floating IPs currently in use.
floating_ips_used = resource.prop("totalFloatingIpsUsed")
floating_ips_used = resource2.Body("totalFloatingIpsUsed")
#: The maximum amount of instances.
instances = resource.prop("maxTotalInstances")
instances = resource2.Body("maxTotalInstances")
#: The amount of instances currently in use.
instances_used = resource.prop("totalInstancesUsed")
instances_used = resource2.Body("totalInstancesUsed")
#: The maximum amount of keypairs.
keypairs = resource.prop("maxTotalKeypairs")
keypairs = resource2.Body("maxTotalKeypairs")
#: The maximum RAM size in megabytes.
total_ram = resource.prop("maxTotalRAMSize")
total_ram = resource2.Body("maxTotalRAMSize")
#: The RAM size in megabytes currently in use.
total_ram_used = resource.prop("totalRAMUsed")
total_ram_used = resource2.Body("totalRAMUsed")
#: The maximum amount of server groups.
server_groups = resource.prop("maxServerGroups")
server_groups = resource2.Body("maxServerGroups")
#: The amount of server groups currently in use.
server_groups_used = resource.prop("totalServerGroupsUsed")
server_groups_used = resource2.Body("totalServerGroupsUsed")
#: The maximum number of members in a server group.
server_group_members = resource.prop("maxServerGroupMembers")
server_group_members = resource2.Body("maxServerGroupMembers")
class RateLimits(resource.Resource):
class RateLimit(resource2.Resource):
#: A list of the specific limits that apply to the ``regex`` and ``uri``.
limits = resource.prop("limit", type=list)
limits = resource2.Body("limit", type=list)
#: A regex representing which routes this rate limit applies to.
regex = resource.prop("regex")
regex = resource2.Body("regex")
#: A URI representing which routes this rate limit applies to.
uri = resource.prop("uri")
uri = resource2.Body("uri")
class Limits(resource.Resource):
class Limits(resource2.Resource):
base_path = "/limits"
resource_key = "limits"
service = compute_service.ComputeService()
allow_retrieve = True
allow_get = True
absolute = resource.prop("absolute", type=AbsoluteLimits)
rate = resource.prop("rate", type=list)
absolute = resource2.Body("absolute", type=AbsoluteLimits)
rate = resource2.Body("rate", type=list)
def get(self, session, args=None, include_headers=False):
def get(self, session):
"""Get the Limits resource.
:param session: The session to use for making this request.
:type session: :class:`~openstack.session.Session`
:param dict args: An optional dict that will be translated into query
strings for retrieving the object when specified.
:returns: A Limits instance
:rtype: :class:`~openstack.compute.v2.limits.Limits`
"""
body = self.get_data_by_id(session, self.id,
include_headers=include_headers)
request = self._prepare_request(requires_id=False, prepend_key=False)
# Split the rates away from absolute limits. We can create
# the `absolute` property and AbsoluteLimits resource directly
# from the body. We have to iterate through the list inside `rate`
# in order to create the RateLimits instances for the `rate` property.
rate_body = body.pop("rate")
self._attrs.update(body)
response = session.get(request.uri, endpoint_filter=self.service)
body = response.json()
body = body[self.resource_key]
absolute_body = self._transpose_component(
body["absolute"], AbsoluteLimits._body_mapping())
self.absolute = AbsoluteLimits.existing(**absolute_body)
rates_body = body["rate"]
rates = []
for rate in rate_body:
rates.append(RateLimits(rate))
for rate_body in rates_body:
rate_body = self._transpose_component(rate_body,
RateLimit._body_mapping())
rates.append(RateLimit(**rate_body))
self.rate = rates
self._attrs.update({"rate": rates})
self._loaded = True
return self

149
openstack/compute/v2/server.py
View File

@ -12,11 +12,11 @@
from openstack.compute import compute_service
from openstack.compute.v2 import metadata
from openstack import resource
from openstack import resource2
from openstack import utils
class Server(resource.Resource, metadata.MetadataMixin):
class Server(resource2.Resource, metadata.MetadataMixin):
resource_key = 'server'
resources_key = 'servers'
base_path = '/servers'
@ -24,89 +24,130 @@ class Server(resource.Resource, metadata.MetadataMixin):
# capabilities
allow_create = True
allow_retrieve = True
allow_get = True
allow_update = True
allow_delete = True
allow_list = True
# Properties
access_ipv4 = resource.prop('accessIPv4')
access_ipv6 = resource.prop('accessIPv6')
_query_mapping = resource2.QueryParameters("image", "flavor", "name",
"status", "host",
changes_since="changes-since")
#: A list of dictionaries holding links relevant to this server.
links = resource2.Body('links')
access_ipv4 = resource2.Body('accessIPv4')
access_ipv6 = resource2.Body('accessIPv6')
#: A dictionary of addresses this server can be accessed through.
#: The dictionary contains keys such as ``private`` and ``public``,
#: each containing a list of dictionaries for addresses of that type.
#: The addresses are contained in a dictionary with keys ``addr``
#: and ``version``, which is either 4 or 6 depending on the protocol
#: of the IP address. *Type: dict*
addresses = resource.prop('addresses', type=dict)
addresses = resource2.Body('addresses', type=dict)
#: Timestamp of when the server was created.
created_at = resource.prop('created')
created_at = resource2.Body('created')
#: The flavor reference, as a ID or full URL, for the flavor to use for
#: this server.
flavor_id = resource.prop('flavorRef')
flavor_id = resource2.Body('flavorRef')
#: An ID representing the host of this server.
host_id = resource.prop('hostId')
host_id = resource2.Body('hostId')
#: The image reference, as a ID or full URL, for the image to use for
#: this server.
image_id = resource.prop('imageRef')
#: A list of dictionaries holding links relevant to this server.
links = resource.prop('links')
image_id = resource2.Body('imageRef')
#: Metadata stored for this server. *Type: dict*
metadata = resource.prop('metadata', type=dict)
#: The name of this server.
name = resource.prop('name')
metadata = resource2.Body('metadata', type=dict)
#: While the server is building, this value represents the percentage
#: of completion. Once it is completed, it will be 100. *Type: int*
progress = resource.prop('progress', type=int)
progress = resource2.Body('progress', type=int)
#: The ID of the project this server is associated with.
project_id = resource.prop('tenant_id')
project_id = resource2.Body('tenant_id')
#: The state this server is in. Valid values include ``ACTIVE``,
#: ``BUILDING``, ``DELETED``, ``ERROR``, ``HARD_REBOOT``, ``PASSWORD``,
#: ``PAUSED``, ``REBOOT``, ``REBUILD``, ``RESCUED``, ``RESIZED``,
#: ``REVERT_RESIZE``, ``SHUTOFF``, ``SOFT_DELETED``, ``STOPPED``,
#: ``SUSPENDED``, ``UNKNOWN``, or ``VERIFY_RESIZE``.
status = resource.prop('status')
status = resource2.Body('status')
#: Timestamp of when this server was last updated.
updated_at = resource.prop('updated')
#: The user ID associated with this server.
user_id = resource.prop('user_id')
updated_at = resource2.Body('updated')
#: The ID of the owners of this server.
user_id = resource2.Body('user_id')
#: The name of an associated keypair
key_name = resource2.Body('key_name')
#: The disk configuration. Either AUTO or MANUAL.
disk_config = resource2.Body('OS-DCF:diskConfig')
#: The name of the availability zone this server is a part of.
availability_zone = resource2.Body('OS-EXT-AZ:availability_zone')
#: The power state of this server.
power_state = resource2.Body('OS-EXT-STS:power_state')
#: The task state of this server.
task_state = resource2.Body('OS-EXT-STS:task_state')
#: The VM state of this server.
vm_state = resource2.Body('OS-EXT-STS:vm_state')
#: A list of an attached volumes. Each item in the list contains at least
#: an "id" key to identify the specific volumes.
attached_volumes = resource2.Body(
'os-extended-volumes:volumes_attached')
#: The timestamp when the server was launched.
launched_at = resource2.Body('OS-SRV-USG:launched_at')
#: The timestamp when the server was terminated (if it has been).
terminated_at = resource2.Body('OS-SRV-USG:terminated_at')
#: A list of applicable security groups. Each group contains keys for
#: description, name, id, and rules.
security_groups = resource2.Body('security_groups')
#: When a server is first created, it provides the administrator password.
admin_password = resource2.Body('adminPass')
#: The file path and contents, text only, to inject into the server at
#: launch. The maximum size of the file path data is 255 bytes.
#: The maximum limit is The number of allowed bytes in the decoded,
#: rather than encoded, data.
personality = resource2.Body('personality')
#: Configuration information or scripts to use upon launch.
#: Must be Base64 encoded.
user_data = resource2.Body('user_data')
#: Enables fine grained control of the block device mapping for an
#: instance. This is typically used for booting servers from volumes.
block_device_mapping = resource2.Body('block_device_mapping_v2', type=dict)
#: The dictionary of data to send to the scheduler.
scheduler_hints = resource2.Body('os:scheduler_hints', type=dict)
#: A networks object. Required parameter when there are multiple
#: networks defined for the tenant. When you do not specify the
#: networks parameter, the server attaches to the only network
#: created for the current tenant.
networks = resource2.Body('networks', type=dict)
@classmethod
def _get_create_body(cls, attrs):
body = {}
if 'scheduler_hints' in attrs:
hints = attrs.pop('scheduler_hints')
body['os:scheduler_hints'] = hints
body[cls.resource_key] = attrs
return body
def action(self, session, body):
def _action(self, session, body):
"""Preform server actions given the message body."""
url = utils.urljoin(self.base_path, self.id, 'action')
# NOTE: This is using Server.base_path instead of self.base_path
# as both Server and ServerDetail instances can be acted on, but
# the URL used is sans any additional /detail/ part.
url = utils.urljoin(Server.base_path, self.id, 'action')
headers = {'Accept': ''}
session.post(
return session.post(
url, endpoint_filter=self.service, json=body, headers=headers)
def change_password(self, session, new_password):
"""Change the administrator password to the given password."""
body = {'changePassword': {'adminPass': new_password}}
return self.action(session, body)
self._action(session, body)
def reboot(self, session, reboot_type):
"""Reboot server where reboot_type might be 'SOFT' or 'HARD'."""
body = {'reboot': {'type': reboot_type}}
return self.action(session, body)
self._action(session, body)
def rebuild(self, session, name, image_href, admin_password,
def rebuild(self, session, name, admin_password,
preserve_ephemeral=False, image=None,
access_ipv4=None, access_ipv6=None,
metadata=None, personality=None):
"""Rebuild the server with the given arguments."""
action = {
'name': name,
'adminPass': admin_password,
'imageRef': image_href,
'preserve_ephemeral': preserve_ephemeral
}
if image is not None:
action['imageRef'] = resource2.Resource._get_id(image)
if access_ipv4 is not None:
action['accessIPv4'] = access_ipv4
if access_ipv6 is not None:
@ -115,23 +156,26 @@ class Server(resource.Resource, metadata.MetadataMixin):
action['metadata'] = metadata
if personality is not None:
action['personality'] = personality
body = {'rebuild': action}
return self.action(session, body)
response = self._action(session, body)
self._translate_response(response)
return self
def resize(self, session, flavor):
"""Resize server to flavor reference."""
body = {'resize': {'flavorRef': flavor}}
return self.action(session, body)
self._action(session, body)
def confirm_resize(self, session):
"""Confirm the resize of the server."""
body = {'confirmResize': None}
return self.action(session, body)
self._action(session, body)
def revert_resize(self, session):
"""Revert the resize of the server."""
body = {'revertResize': None}
return self.action(session, body)
self._action(session, body)
def create_image(self, session, name, metadata=None):
"""Create image from server."""
@ -139,16 +183,15 @@ class Server(resource.Resource, metadata.MetadataMixin):
if metadata is not None:
action['metadata'] = metadata
body = {'createImage': action}
return self.action(session, body)
self._action(session, body)
def get_floating_ips(self):
"""Get the floating ips associated with this server."""
addresses = self.addresses[self.name]
result = []
for address in addresses:
if address['OS-EXT-IPS:type'] == 'floating':
result.append(address['addr'])
return result
def add_security_group(self, session, security_group):
body = {"addSecurityGroup": {"name": security_group}}
self._action(session, body)
def remove_security_group(self, session, security_group):
body = {"removeSecurityGroup": {"name": security_group}}
self._action(session, body)
class ServerDetail(Server):
@ -156,7 +199,7 @@ class ServerDetail(Server):
# capabilities
allow_create = False
allow_retrieve = False
allow_get = False
allow_update = False
allow_delete = False
allow_list = True

16
openstack/compute/v2/server_group.py
View File

@ -11,27 +11,29 @@
# under the License.
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class ServerGroup(resource.Resource):
class ServerGroup(resource2.Resource):
resource_key = 'server_group'
resources_key = 'server_groups'
base_path = '/os-server-groups'
service = compute_service.ComputeService()
_query_mapping = resource2.QueryParameters("all_projects")
# capabilities
allow_create = True
allow_retrieve = True
allow_get = True
allow_delete = True
allow_list = True
# Properties
#: A name identifying the server group
name = resource.prop('name')
name = resource2.Body('name')
#: The list of policies supported by the server group
policies = resource.prop('policies')
policies = resource2.Body('policies')
#: The list of members in the server group
member_ids = resource.prop('members')
member_ids = resource2.Body('members')
#: The metadata associated with the server group
metadata = resource.prop('metadata')
metadata = resource2.Body('metadata')

20
openstack/compute/v2/server_interface.py
View File

@ -11,11 +11,10 @@
# under the License.
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
class ServerInterface(resource.Resource):
id_attribute = 'port_id'
class ServerInterface(resource2.Resource):
resource_key = 'interfaceAttachment'
resources_key = 'interfaceAttachments'
base_path = '/servers/%(server_id)s/os-interface'
@ -23,21 +22,20 @@ class ServerInterface(resource.Resource):
# capabilities
allow_create = True
allow_retrieve = True
allow_get = True
allow_update = False
allow_delete = True
allow_list = True
# Properties
#: Fixed IP addresses with subnet IDs.
fixed_ips = resource.prop('fixed_ips')
fixed_ips = resource2.Body('fixed_ips')
#: The MAC address.
mac_addr = resource.prop('mac_addr')
mac_addr = resource2.Body('mac_addr')
#: The network ID.
net_id = resource.prop('net_id')
net_id = resource2.Body('net_id')
#: The ID of the port for which you want to create an interface.
port_id = resource.prop('port_id')
port_id = resource2.Body('port_id', alternate_id=True)
#: The port state.
port_state = resource.prop('port_state')
port_state = resource2.Body('port_state')
#: The ID for the server.
server_id = resource.prop('server_id')
server_id = resource2.URI('server_id')

48
openstack/compute/v2/server_ip.py
View File

@ -10,16 +10,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack.compute import compute_service
from openstack import resource
from openstack import resource2
from openstack import utils
class ServerIP(resource.Resource):
id_attribute = 'addr'
resource_key = 'server_ip'
resources_key = 'server_ips'
class ServerIP(resource2.Resource):
resources_key = 'addresses'
base_path = '/servers/%(server_id)s/ips'
service = compute_service.ComputeService()
@ -28,27 +25,30 @@ class ServerIP(resource.Resource):
# Properties
#: The IP address. The format of the address depends on :attr:`version`
addr = resource.prop('addr')
address = resource2.Body('addr')
#: The network label, such as public or private.
network_label = resource.prop('network_label')
network_label = resource2.URI('network_label')
#: The ID for the server.
server_id = resource.prop('server_id')
server_id = resource2.URI('server_id')
# Version of the IP protocol. Currently either 4 or 6.
version = resource.prop('version')
version = resource2.Body('version')
@classmethod
def list(cls, session, path_args=None, **params):
url = cls._get_url(path_args)
resp = session.get(url, endpoint_filter=cls.service, params=params)
def list(cls, session, paginated=False, server_id=None,
network_label=None, **params):
url = cls.base_path % {"server_id": server_id}
if network_label is not None:
url = utils.urljoin(url, network_label)
resp = session.get(url, endpoint_filter=cls.service)
resp = resp.json()
ray = []
for network_label, addresses in six.iteritems(resp['addresses']):
if network_label is None:
resp = resp[cls.resources_key]
for label, addresses in resp.items():
for address in addresses:
record = {
'server_id': path_args['server_id'],
'network_label': network_label,
'version': address['version'],
'addr': address['addr'],
}
ray.append(cls.existing(**record))
return ray
yield cls.existing(network_label=label,
address=address["addr"],
version=address["version"])

12
openstack/tests/functional/compute/v2/test_image.py
View File

@ -46,12 +46,12 @@ class TestImage(base.BaseFunctionalTest):
sot = self.conn.compute.get_image(image.id)
self.assertEqual(image.id, sot.id)
self.assertEqual(image.name, sot.name)
self.assertIn('links', image)
self.assertIn('minDisk', image)
self.assertIn('minRam', image)
self.assertIn('metadata', image)
self.assertIn('progress', image)
self.assertIn('status', image)
self.assertIsNotNone(image.links)
self.assertIsNotNone(image.min_disk)
self.assertIsNotNone(image.min_ram)
self.assertIsNotNone(image.metadata)
self.assertIsNotNone(image.progress)
self.assertIsNotNone(image.status)
def test_image_metadata(self):
image = self._get_non_test_image()

3
openstack/tests/functional/compute/v2/test_keypair.py
View File

@ -27,11 +27,12 @@ class TestKeypair(base.BaseFunctionalTest):
sot = cls.conn.compute.create_keypair(name=cls.NAME)
assert isinstance(sot, keypair.Keypair)
cls.assertIs(cls.NAME, sot.name)
cls._keypair = sot
cls.ID = sot.id
@classmethod
def tearDownClass(cls):
sot = cls.conn.compute.delete_keypair(cls.ID)
sot = cls.conn.compute.delete_keypair(cls._keypair)
cls.assertIs(None, sot)
def test_find(self):

10
openstack/tests/functional/compute/v2/test_limits.py
View File

@ -17,8 +17,8 @@ class TestLimits(base.BaseFunctionalTest):
def test_limits(self):
sot = self.conn.compute.get_limits()
self.assertIn('maxTotalInstances', sot.absolute)
self.assertIn('maxTotalRAMSize', sot.absolute)
self.assertIn('maxTotalKeypairs', sot.absolute)
self.assertIn('maxSecurityGroups', sot.absolute)
self.assertIn('maxSecurityGroupRules', sot.absolute)
self.assertIsNotNone('maxTotalInstances', sot.absolute)
self.assertIsNotNone('maxTotalRAMSize', sot.absolute)
self.assertIsNotNone('maxTotalKeypairs', sot.absolute)
self.assertIsNotNone('maxSecurityGroups', sot.absolute)
self.assertIsNotNone('maxSecurityGroupRules', sot.absolute)

13
openstack/tests/functional/compute/v2/test_server.py
View File

@ -35,12 +35,15 @@ class TestServer(base.BaseFunctionalTest):
cls.network, cls.subnet = test_network.create_network(cls.conn,
cls.NAME,
cls.cidr)
if cls.network:
args = {'networks': [{"uuid": cls.network.id}]}
else:
args = {}
if not cls.network:
# We can't call TestCase.fail from within the setUpClass
# classmethod, but we need to raise some exception in order
# to get this setup to fail and thusly fail the entire class.
raise Exception("Unable to create network for TestServer")
sot = cls.conn.compute.create_server(
name=cls.NAME, flavor_id=flavor.id, image_id=image.id, **args)
name=cls.NAME, flavor_id=flavor.id, image_id=image.id,
networks=[{"uuid": cls.network.id}])
cls.conn.compute.wait_for_server(sot)
assert isinstance(sot, server.Server)
cls.assertIs(cls.NAME, sot.name)

10
openstack/tests/unit/compute/v2/test_availability_zone.py
View File

@ -27,14 +27,20 @@ class TestAvailabilityZone(testtools.TestCase):
def test_basic(self):
sot = az.AvailabilityZone()
self.assertEqual('availability_zone', sot.resource_key)
self.assertEqual('availabilityZoneInfo', sot.resources_key)
self.assertEqual('/os-availability-zone', sot.base_path)
self.assertTrue(sot.allow_list)
self.assertEqual('compute', sot.service.service_type)
def test_basic_detail(self):
sot = az.AvailabilityZoneDetail()
self.assertEqual('availabilityZoneInfo', sot.resources_key)
self.assertEqual('/os-availability-zone/detail', sot.base_path)
self.assertTrue(sot.allow_list)
self.assertEqual('compute', sot.service.service_type)
def test_make_basic(self):
sot = az.AvailabilityZone(BASIC_EXAMPLE)
sot = az.AvailabilityZone(**BASIC_EXAMPLE)
self.assertEqual(BASIC_EXAMPLE['id'], sot.id)
self.assertEqual(BASIC_EXAMPLE['zoneState'], sot.state)
self.assertEqual(BASIC_EXAMPLE['hosts'], sot.hosts)

5
openstack/tests/unit/compute/v2/test_extension.py
View File

@ -34,14 +34,13 @@ class TestExtension(testtools.TestCase):
self.assertEqual('/extensions', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertFalse(sot.allow_create)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertTrue(sot.allow_list)
def test_make_it(self):
sot = extension.Extension(EXAMPLE)
self.assertEqual(EXAMPLE['alias'], sot.id)
sot = extension.Extension(**EXAMPLE)
self.assertEqual(EXAMPLE['alias'], sot.alias)
self.assertEqual(EXAMPLE['description'], sot.description)
self.assertEqual(EXAMPLE['links'], sot.links)

12
openstack/tests/unit/compute/v2/test_flavor.py
View File

@ -39,13 +39,19 @@ class TestFlavor(testtools.TestCase):
self.assertEqual('/flavors', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertTrue(sot.allow_create)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertTrue(sot.allow_update)
self.assertTrue(sot.allow_delete)
self.assertTrue(sot.allow_list)
self.assertDictEqual({"sort_key": "sort_key",
"sort_dir": "sort_dir",
"min_disk": "minDisk",
"min_ram": "minRam"},
sot._query_mapping._mapping)
def test_make_basic(self):
sot = flavor.Flavor(BASIC_EXAMPLE)
sot = flavor.Flavor(**BASIC_EXAMPLE)
self.assertEqual(BASIC_EXAMPLE['id'], sot.id)
self.assertEqual(BASIC_EXAMPLE['links'], sot.links)
self.assertEqual(BASIC_EXAMPLE['name'], sot.name)
@ -68,7 +74,7 @@ class TestFlavor(testtools.TestCase):
self.assertEqual('/flavors/detail', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertFalse(sot.allow_create)
self.assertFalse(sot.allow_retrieve)
self.assertFalse(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertTrue(sot.allow_list)

51
openstack/tests/unit/compute/v2/test_hypervisor.py
View File

@ -15,10 +15,30 @@ import testtools
from openstack.compute.v2 import hypervisor
EXAMPLE = {
'id': 'IDENTIFIER',
'name': 'hypervisor_hostname',
'state': 'up',
'status': 'enabled',
"status": "enabled",
"service": {
"host": "fake-mini",
"disabled_reason": None,
"id": 6
},
"vcpus_used": 0,
"hypervisor_type": "QEMU",
"local_gb_used": 0,
"vcpus": 8,
"hypervisor_hostname": "fake-mini",
"memory_mb_used": 512,
"memory_mb": 7980,
"current_workload": 0,
"state": "up",
"host_ip": "23.253.248.171",
"cpu_info": "some cpu info",
"running_vms": 0,
"free_disk_gb": 157,
"hypervisor_version": 2000000,
"disk_available_least": 140,
"local_gb": 157,
"free_ram_mb": 7468,
"id": 1
}
@ -30,12 +50,29 @@ class TestHypervisor(testtools.TestCase):
self.assertEqual('hypervisors', sot.resources_key)
self.assertEqual('/os-hypervisors', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertTrue(sot.allow_list)
def test_make_it(self):
sot = hypervisor.Hypervisor(EXAMPLE)
sot = hypervisor.Hypervisor(**EXAMPLE)
self.assertEqual(EXAMPLE['id'], sot.id)
self.assertEqual(EXAMPLE['name'], sot.name)
self.assertEqual(EXAMPLE['hypervisor_hostname'], sot.name)
self.assertEqual(EXAMPLE['state'], sot.state)
self.assertEqual(EXAMPLE['status'], sot.status)
self.assertEqual(EXAMPLE['service'], sot.service_details)
self.assertEqual(EXAMPLE['vcpus_used'], sot.vcpus_used)
self.assertEqual(EXAMPLE['hypervisor_type'], sot.hypervisor_type)
self.assertEqual(EXAMPLE['local_gb_used'], sot.local_disk_used)
self.assertEqual(EXAMPLE['vcpus'], sot.vcpus)
self.assertEqual(EXAMPLE['vcpus_used'], sot.vcpus_used)
self.assertEqual(EXAMPLE['memory_mb_used'], sot.memory_used)
self.assertEqual(EXAMPLE['memory_mb'], sot.memory_size)
self.assertEqual(EXAMPLE['current_workload'], sot.current_workload)
self.assertEqual(EXAMPLE['host_ip'], sot.host_ip)
self.assertEqual(EXAMPLE['cpu_info'], sot.cpu_info)
self.assertEqual(EXAMPLE['running_vms'], sot.running_vms)
self.assertEqual(EXAMPLE['free_disk_gb'], sot.local_disk_free)
self.assertEqual(EXAMPLE['hypervisor_version'], sot.hypervisor_version)
self.assertEqual(EXAMPLE['disk_available_least'], sot.disk_available)
self.assertEqual(EXAMPLE['local_gb'], sot.local_disk_size)
self.assertEqual(EXAMPLE['free_ram_mb'], sot.memory_free)

17
openstack/tests/unit/compute/v2/test_image.py
View File

@ -45,13 +45,22 @@ class TestImage(testtools.TestCase):
self.assertEqual('/images', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertFalse(sot.allow_create)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertTrue(sot.allow_delete)
self.assertTrue(sot.allow_list)
self.assertDictEqual({"server": "server",
"name": "name",
"status": "status",
"type": "type",
"min_disk": "minDisk",
"min_ram": "minRam",
"changes_since": "changes-since"},
sot._query_mapping._mapping)
def test_make_basic(self):
sot = image.Image(BASIC_EXAMPLE)
sot = image.Image(**BASIC_EXAMPLE)
self.assertEqual(BASIC_EXAMPLE['id'], sot.id)
self.assertEqual(BASIC_EXAMPLE['links'], sot.links)
self.assertEqual(BASIC_EXAMPLE['name'], sot.name)
@ -63,13 +72,13 @@ class TestImage(testtools.TestCase):
self.assertEqual('/images/detail', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertFalse(sot.allow_create)
self.assertFalse(sot.allow_retrieve)
self.assertFalse(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertTrue(sot.allow_list)
def test_make_detail(self):
sot = image.ImageDetail(DETAIL_EXAMPLE)
sot = image.ImageDetail(**DETAIL_EXAMPLE)
self.assertEqual(DETAIL_EXAMPLE['created'], sot.created_at)
self.assertEqual(DETAIL_EXAMPLE['id'], sot.id)
self.assertEqual(DETAIL_EXAMPLE['links'], sot.links)

22
openstack/tests/unit/compute/v2/test_keypair.py
View File

@ -15,11 +15,10 @@ import testtools
from openstack.compute.v2 import keypair
EXAMPLE = {
'keypair': {
'fingerprint': '1',
'name': '2',
'public_key': '3',
}
'fingerprint': '1',
'name': '2',
'public_key': '3',
'private_key': '3',
}
@ -32,13 +31,14 @@ class TestKeypair(testtools.TestCase):
self.assertEqual('/os-keypairs', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertTrue(sot.allow_create)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_update)
self.assertTrue(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertTrue(sot.allow_delete)
self.assertTrue(sot.allow_list)
def test_make_it(self):
sot = keypair.Keypair(EXAMPLE)
self.assertEqual(EXAMPLE['keypair']['fingerprint'], sot.fingerprint)
self.assertEqual(EXAMPLE['keypair']['name'], sot.name)
self.assertEqual(EXAMPLE['keypair']['public_key'], sot.public_key)
sot = keypair.Keypair(**EXAMPLE)
self.assertEqual(EXAMPLE['fingerprint'], sot.fingerprint)
self.assertEqual(EXAMPLE['name'], sot.name)
self.assertEqual(EXAMPLE['public_key'], sot.public_key)
self.assertEqual(EXAMPLE['private_key'], sot.private_key)

70
openstack/tests/unit/compute/v2/test_limits.py
View File

@ -68,13 +68,13 @@ class TestAbsoluteLimits(testtools.TestCase):
self.assertEqual("", sot.base_path)
self.assertIsNone(sot.service)
self.assertFalse(sot.allow_create)
self.assertFalse(sot.allow_retrieve)
self.assertFalse(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertFalse(sot.allow_list)
def test_make_it(self):
sot = limits.AbsoluteLimits(ABSOLUTE_LIMITS)
sot = limits.AbsoluteLimits(**ABSOLUTE_LIMITS)
self.assertEqual(ABSOLUTE_LIMITS["maxImageMeta"], sot.image_meta)
self.assertEqual(ABSOLUTE_LIMITS["maxPersonality"], sot.personality)
self.assertEqual(ABSOLUTE_LIMITS["maxPersonalitySize"],
@ -109,22 +109,22 @@ class TestAbsoluteLimits(testtools.TestCase):
sot.total_cores_used)
class TestRateLimits(testtools.TestCase):
class TestRateLimit(testtools.TestCase):
def test_basic(self):
sot = limits.RateLimits()
sot = limits.RateLimit()
self.assertIsNone(sot.resource_key)
self.assertIsNone(sot.resources_key)
self.assertEqual("", sot.base_path)
self.assertIsNone(sot.service)
self.assertFalse(sot.allow_create)
self.assertFalse(sot.allow_retrieve)
self.assertFalse(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertFalse(sot.allow_list)
def test_make_it(self):
sot = limits.RateLimits(RATE_LIMIT)
sot = limits.RateLimit(**RATE_LIMIT)
self.assertEqual(RATE_LIMIT["regex"], sot.regex)
self.assertEqual(RATE_LIMIT["uri"], sot.uri)
self.assertEqual(RATE_LIMIT["limit"], sot.limits)
@ -137,19 +137,59 @@ class TestLimits(testtools.TestCase):
self.assertEqual("limits", sot.resource_key)
self.assertEqual("/limits", sot.base_path)
self.assertEqual("compute", sot.service.service_type)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertFalse(sot.allow_create)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertFalse(sot.allow_list)
@mock.patch("openstack.resource.Resource.get_data_by_id")
def test_get(self, mock_get):
# Only return values under the limits key since that's our
# resource_key, which would be filtered out in get_data_by_id.
mock_get.return_value = LIMITS_BODY["limits"]
def test_get(self):
sess = mock.Mock()
resp = mock.Mock()
sess.get.return_value = resp
resp.json.return_value = LIMITS_BODY
sot = limits.Limits().get("fake session")
sot = limits.Limits().get(sess)
self.assertEqual(sot.absolute, limits.AbsoluteLimits(ABSOLUTE_LIMITS))
self.assertEqual(sot.rate, [limits.RateLimits(RATE_LIMIT)])
self.assertEqual(ABSOLUTE_LIMITS["maxImageMeta"],
sot.absolute.image_meta)
self.assertEqual(ABSOLUTE_LIMITS["maxPersonality"],
sot.absolute.personality)
self.assertEqual(ABSOLUTE_LIMITS["maxPersonalitySize"],
sot.absolute.personality_size)
self.assertEqual(ABSOLUTE_LIMITS["maxSecurityGroupRules"],
sot.absolute.security_group_rules)
self.assertEqual(ABSOLUTE_LIMITS["maxSecurityGroups"],
sot.absolute.security_groups)
self.assertEqual(ABSOLUTE_LIMITS["maxServerMeta"],
sot.absolute.server_meta)
self.assertEqual(ABSOLUTE_LIMITS["maxTotalCores"],
sot.absolute.total_cores)
self.assertEqual(ABSOLUTE_LIMITS["maxTotalFloatingIps"],
sot.absolute.floating_ips)
self.assertEqual(ABSOLUTE_LIMITS["maxTotalInstances"],
sot.absolute.instances)
self.assertEqual(ABSOLUTE_LIMITS["maxTotalKeypairs"],
sot.absolute.keypairs)
self.assertEqual(ABSOLUTE_LIMITS["maxTotalRAMSize"],
sot.absolute.total_ram)
self.assertEqual(ABSOLUTE_LIMITS["maxServerGroups"],
sot.absolute.server_groups)
self.assertEqual(ABSOLUTE_LIMITS["maxServerGroupMembers"],
sot.absolute.server_group_members)
self.assertEqual(ABSOLUTE_LIMITS["totalFloatingIpsUsed"],
sot.absolute.floating_ips_used)
self.assertEqual(ABSOLUTE_LIMITS["totalSecurityGroupsUsed"],
sot.absolute.security_groups_used)
self.assertEqual(ABSOLUTE_LIMITS["totalRAMUsed"],
sot.absolute.total_ram_used)
self.assertEqual(ABSOLUTE_LIMITS["totalInstancesUsed"],
sot.absolute.instances_used)
self.assertEqual(ABSOLUTE_LIMITS["totalServerGroupsUsed"],
sot.absolute.server_groups_used)
self.assertEqual(ABSOLUTE_LIMITS["totalCoresUsed"],
sot.absolute.total_cores_used)
self.assertEqual(RATE_LIMIT["uri"], sot.rate[0].uri)
self.assertEqual(RATE_LIMIT["regex"], sot.rate[0].regex)
self.assertEqual(RATE_LIMIT["limit"], sot.rate[0].limits)

8
openstack/tests/unit/compute/v2/test_metadata.py
View File

@ -32,12 +32,12 @@ class TestMetadata(testtools.TestCase):
self.meta_result = {"meta": {"oh": "yeah"}}
def test_get_all_metadata_Server(self):
self._test_get_all_metadata(server.Server({"id": IDENTIFIER}))
self._test_get_all_metadata(server.Server(id=IDENTIFIER))
def test_get_all_metadata_ServerDetail(self):
# This is tested explicitly so we know ServerDetail items are
# properly having /detail stripped out of their base_path.
self._test_get_all_metadata(server.ServerDetail({"id": IDENTIFIER}))
self._test_get_all_metadata(server.ServerDetail(id=IDENTIFIER))
def _test_get_all_metadata(self, sot):
response = mock.Mock()
@ -58,7 +58,7 @@ class TestMetadata(testtools.TestCase):
sess = mock.Mock()
sess.post.return_value = response
sot = server.Server({"id": IDENTIFIER})
sot = server.Server(id=IDENTIFIER)
set_meta = {"lol": "rofl"}
@ -74,7 +74,7 @@ class TestMetadata(testtools.TestCase):
sess = mock.Mock()
sess.delete.return_value = None
sot = server.Server({"id": IDENTIFIER})
sot = server.Server(id=IDENTIFIER)
key = "hey"

190
openstack/tests/unit/compute/v2/test_proxy.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from openstack.compute.v2 import _proxy
from openstack.compute.v2 import availability_zone as az
from openstack.compute.v2 import extension
@ -24,10 +22,10 @@ from openstack.compute.v2 import server
from openstack.compute.v2 import server_group
from openstack.compute.v2 import server_interface
from openstack.compute.v2 import server_ip
from openstack.tests.unit import test_proxy_base
from openstack.tests.unit import test_proxy_base2
class TestComputeProxy(test_proxy_base.TestProxyBase):
class TestComputeProxy(test_proxy_base2.TestProxyBase):
def setUp(self):
super(TestComputeProxy, self).setUp()
self.proxy = _proxy.Proxy(self.session)
@ -36,8 +34,8 @@ class TestComputeProxy(test_proxy_base.TestProxyBase):
self.verify_find(self.proxy.find_extension, extension.Extension)
def test_extensions(self):
self.verify_list(self.proxy.extensions, extension.Extension,
paginated=False)
self.verify_list_no_kwargs(self.proxy.extensions, extension.Extension,
paginated=False)
def test_flavor_create(self):
self.verify_create(self.proxy.create_flavor, flavor.Flavor)
@ -109,11 +107,8 @@ class TestComputeProxy(test_proxy_base.TestProxyBase):
self.verify_get(self.proxy.get_keypair, keypair.Keypair)
def test_keypairs(self):
self.verify_list(self.proxy.keypairs, keypair.Keypair,
paginated=False)
def test_keypair_update(self):
self.verify_update(self.proxy.update_keypair, keypair.Keypair)
self.verify_list_no_kwargs(self.proxy.keypairs, keypair.Keypair,
paginated=False)
def test_limits_get(self):
self.verify_get(self.proxy.get_limits, limits.Limits, value=[])
@ -122,75 +117,89 @@ class TestComputeProxy(test_proxy_base.TestProxyBase):
self.verify_create(self.proxy.create_server_interface,
server_interface.ServerInterface,
method_kwargs={"server": "test_id"},
expected_kwargs={"path_args": {
"server_id": "test_id"}})
expected_kwargs={"server_id": "test_id"})
def test_server_interface_delete(self):
test_interface = server_interface.ServerInterface.from_id(
"test_interface_id")
test_interface.server_id = "test_server_id"
self.proxy._get_uri_attribute = lambda *args: args[1]
interface_id = "test_interface_id"
server_id = "test_server_id"
test_interface = server_interface.ServerInterface(id=interface_id)
test_interface.server_id = server_id
# Case1: ServerInterface instance is provided as value
self._verify2("openstack.proxy.BaseProxy._delete",
self._verify2("openstack.proxy2.BaseProxy._delete",
self.proxy.delete_server_interface,
method_args=[test_interface, "test_server_id"],
expected_args=[server_interface.ServerInterface,
test_interface],
expected_kwargs={"path_args": {
"server_id": "test_server_id"},
"ignore_missing": True})
method_args=[test_interface],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"server_id": server_id,
"port_id": interface_id,
"ignore_missing": True})
# Case2: ServerInterface ID is provided as value
self._verify2("openstack.proxy.BaseProxy._delete",
self._verify2("openstack.proxy2.BaseProxy._delete",
self.proxy.delete_server_interface,
method_args=["test_interface_id", "test_server_id"],
expected_args=[server_interface.ServerInterface,
"test_interface_id"],
expected_kwargs={"path_args": {
"server_id": "test_server_id"},
"ignore_missing": True})
method_args=[interface_id],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"server_id": server_id,
"port_id": interface_id,
"ignore_missing": True})
def test_server_interface_delete_ignore(self):
self.proxy._get_uri_attribute = lambda *args: args[1]
self.verify_delete(self.proxy.delete_server_interface,
server_interface.ServerInterface, True,
{"server": "test_id"}, {"server_id": "test_id"})
method_kwargs={"server": "test_id"},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"server_id": "test_id",
"port_id": "resource_or_id"})
def test_server_interface_get(self):
test_interface = server_interface.ServerInterface.from_id(
"test_interface_id")
test_interface.server_id = "test_server_id"
self.proxy._get_uri_attribute = lambda *args: args[1]
interface_id = "test_interface_id"
server_id = "test_server_id"
test_interface = server_interface.ServerInterface(id=interface_id)
test_interface.server_id = server_id
# Case1: ServerInterface instance is provided as value
self._verify2('openstack.proxy.BaseProxy._get',
self._verify2('openstack.proxy2.BaseProxy._get',
self.proxy.get_server_interface,
method_args=[test_interface, "test_id"],
expected_args=[server_interface.ServerInterface,
test_interface],
expected_kwargs={"path_args": {
"server_id": "test_server_id"}})
method_args=[test_interface],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"port_id": interface_id,
"server_id": server_id})
# Case2: ServerInterface ID is provided as value
self._verify2('openstack.proxy.BaseProxy._get',
self._verify2('openstack.proxy2.BaseProxy._get',
self.proxy.get_server_interface,
method_args=["test_interface_id", "test_server_id"],
expected_args=[server_interface.ServerInterface,
"test_interface_id"],
expected_kwargs={"path_args": {
"server_id": "test_server_id"}})
method_args=[interface_id],
method_kwargs={"server": server_id},
expected_args=[server_interface.ServerInterface],
expected_kwargs={"port_id": interface_id,
"server_id": server_id})
def test_server_interfaces(self):
self.verify_list(self.proxy.server_interfaces,
server_interface.ServerInterface,
paginated=False, method_args=["test_id"],
expected_kwargs={"path_args": {
"server_id": "test_id"}})
expected_kwargs={"server_id": "test_id"})
def test_server_ip_find(self):
self.verify_find(self.proxy.find_server_ip, server_ip.ServerIP)
def test_server_ips(self):
def test_server_ips_with_network_label(self):
self.verify_list(self.proxy.server_ips, server_ip.ServerIP,
paginated=False)
paginated=False, method_args=["test_id"],
method_kwargs={"network_label": "test_label"},
expected_kwargs={"server_id": "test_id",
"network_label": "test_label"})
def test_server_ips_without_network_label(self):
self.verify_list(self.proxy.server_ips, server_ip.ServerIP,
paginated=False, method_args=["test_id"],
expected_kwargs={"server_id": "test_id",
"network_label": None})
def test_server_create_attrs(self):
self.verify_create(self.proxy.create_server, server.Server)
@ -212,7 +221,7 @@ class TestComputeProxy(test_proxy_base.TestProxyBase):
paginated=True,
method_kwargs={"details": True,
"changes_since": 1, "image": 2},
expected_kwargs={"changes-since": 1, "image": 2})
expected_kwargs={"changes_since": 1, "image": 2})
def test_servers_not_detailed(self):
self.verify_list(self.proxy.servers, server.Server,
@ -220,13 +229,13 @@ class TestComputeProxy(test_proxy_base.TestProxyBase):
method_kwargs={"details": False,
"changes_since": 1, "image": 2},
expected_kwargs={"paginated": True,
"changes-since": 1, "image": 2})
"changes_since": 1, "image": 2})
def test_server_update(self):
self.verify_update(self.proxy.update_server, server.Server)
def test_server_wait_for(self):
value = server.Server(attrs={'id': '1234'})
value = server.Server(id='1234')
self.verify_wait_for_status(
self.proxy.wait_for_server,
method_args=[value],
@ -240,69 +249,61 @@ class TestComputeProxy(test_proxy_base.TestProxyBase):
def test_server_confirm_resize(self):
self._verify("openstack.compute.v2.server.Server.confirm_resize",
self.proxy.confirm_resize_server,
self.proxy.confirm_server_resize,
method_args=["value"])
def test_server_revert_resize(self):
self._verify("openstack.compute.v2.server.Server.revert_resize",
self.proxy.revert_resize_server,
self.proxy.revert_server_resize,
method_args=["value"])
@mock.patch.object(_proxy.Proxy, 'find_image')
def test_server_rebuild(self, mock_find_image):
image_obj = image.Image.from_id('test_image_id')
mock_find_image.side_effect = [image_obj, None]
def test_server_rebuild(self):
id = 'test_image_id'
image_obj = image.Image(id='test_image_id')
# Case1: image object is provided as image_ref
# Case1: image object is provided
# NOTE: Inside of Server.rebuild is where image_obj gets converted
# to an ID instead of object.
self._verify('openstack.compute.v2.server.Server.rebuild',
self.proxy.rebuild_server,
method_args=["value", image_obj, "test_server",
"test_pass"],
method_kwargs={"metadata": {"k1": "v1"}},
expected_args=["test_server", "test_image_id",
"test_pass"],
expected_kwargs={"metadata": {"k1": "v1"}})
method_args=["value", "test_server", "test_pass"],
method_kwargs={"metadata": {"k1": "v1"},
"image": image_obj},
expected_args=["test_server", "test_pass"],
expected_kwargs={"metadata": {"k1": "v1"},
"image": image_obj})
# Case2: image name or id is provided as image_ref
# Case2: image name or id is provided
self._verify('openstack.compute.v2.server.Server.rebuild',
self.proxy.rebuild_server,
method_args=["value", "test_image_name_or_id",
"test_server", "test_pass"],
method_kwargs={"metadata": {"k1": "v1"}},
expected_args=["test_server", "test_image_id",
"test_pass"],
expected_kwargs={"metadata": {"k1": "v1"}})
# Case3: image URL is provided as image_ref
self._verify('openstack.compute.v2.server.Server.rebuild',
self.proxy.rebuild_server,
method_args=["value", "test_image_url", "test_server",
"test_pass"],
method_kwargs={"metadata": {"k1": "v1"}},
expected_args=["test_server", "test_image_url",
"test_pass"],
expected_kwargs={"metadata": {"k1": "v1"}})
method_args=["value", "test_server", "test_pass"],
method_kwargs={"metadata": {"k1": "v1"},
"image": id},
expected_args=["test_server", "test_pass"],
expected_kwargs={"metadata": {"k1": "v1"},
"image": id})
def test_availability_zones(self):
self.verify_list(self.proxy.availability_zones, az.AvailabilityZone,
paginated=False)
self.verify_list_no_kwargs(self.proxy.availability_zones,
az.AvailabilityZone,
paginated=False)
def test_get_all_server_metadata(self):
self._verify2("openstack.compute.v2.server.Server.get_metadata",
self.proxy.get_server_metadata,
method_args=["value"],
method_result=server.Server.existing(id="value",
metadata={}),
method_result=server.Server(id="value", metadata={}),
expected_args=[self.session],
expected_result={})
def test_set_server_metadata(self):
kwargs = {"a": "1", "b": "2"}
id = "an_id"
self._verify2("openstack.compute.v2.server.Server.set_metadata",
self.proxy.set_server_metadata,
method_args=["value"],
method_args=[id],
method_kwargs=kwargs,
method_result=server.Server.existing(id="value",
method_result=server.Server.existing(id=id,
metadata=kwargs),
expected_args=[self.session],
expected_kwargs=kwargs,
@ -340,8 +341,9 @@ class TestComputeProxy(test_proxy_base.TestProxyBase):
paginated=False)
def test_hypervisors(self):
self.verify_list(self.proxy.hypervisors, hypervisor.Hypervisor,
paginated=False)
self.verify_list_no_kwargs(self.proxy.hypervisors,
hypervisor.Hypervisor,
paginated=False)
def test_find_hypervisor(self):
self.verify_find(self.proxy.find_hypervisor,

186
openstack/tests/unit/compute/v2/test_server.py
View File

@ -10,8 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import mock
import testtools
@ -35,6 +33,21 @@ EXAMPLE = {
'status': '14',
'updated': '2015-03-09T12:15:57.233772',
'user_id': '16',
'key_name': '17',
'OS-DCF:diskConfig': '18',
'OS-EXT-AZ:availability_zone': '19',
'OS-EXT-STS:power_state': '20',
'OS-EXT-STS:task_state': '21',
'OS-EXT-STS:vm_state': '22',
'os-extended-volumes:volumes_attached': '23',
'OS-SRV-USG:launched_at': '2015-03-09T12:15:57.233772',
'OS-SRV-USG:terminated_at': '2015-03-09T12:15:57.233772',
'security_groups': '26',
'adminPass': '27',
'personality': '28',
'block_device_mapping_v2': {'key': '29'},
'os:scheduler_hints': {'key': '30'},
'user_data': '31'
}
@ -55,13 +68,21 @@ class TestServer(testtools.TestCase):
self.assertEqual('/servers', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertTrue(sot.allow_create)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertTrue(sot.allow_update)
self.assertTrue(sot.allow_delete)
self.assertTrue(sot.allow_list)
self.assertDictEqual({"image": "image",
"flavor": "flavor",
"name": "name",
"status": "status",
"host": "host",
"changes_since": "changes-since"},
sot._query_mapping._mapping)
def test_make_it(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
self.assertEqual(EXAMPLE['accessIPv4'], sot.access_ipv4)
self.assertEqual(EXAMPLE['accessIPv6'], sot.access_ipv6)
self.assertEqual(EXAMPLE['addresses'], sot.addresses)
@ -78,6 +99,25 @@ class TestServer(testtools.TestCase):
self.assertEqual(EXAMPLE['status'], sot.status)
self.assertEqual(EXAMPLE['updated'], sot.updated_at)
self.assertEqual(EXAMPLE['user_id'], sot.user_id)
self.assertEqual(EXAMPLE['key_name'], sot.key_name)
self.assertEqual(EXAMPLE['OS-DCF:diskConfig'], sot.disk_config)
self.assertEqual(EXAMPLE['OS-EXT-AZ:availability_zone'],
sot.availability_zone)
self.assertEqual(EXAMPLE['OS-EXT-STS:power_state'], sot.power_state)
self.assertEqual(EXAMPLE['OS-EXT-STS:task_state'], sot.task_state)
self.assertEqual(EXAMPLE['OS-EXT-STS:vm_state'], sot.vm_state)
self.assertEqual(EXAMPLE['os-extended-volumes:volumes_attached'],
sot.attached_volumes)
self.assertEqual(EXAMPLE['OS-SRV-USG:launched_at'], sot.launched_at)
self.assertEqual(EXAMPLE['OS-SRV-USG:terminated_at'],
sot.terminated_at)
self.assertEqual(EXAMPLE['security_groups'], sot.security_groups)
self.assertEqual(EXAMPLE['adminPass'], sot.admin_password)
self.assertEqual(EXAMPLE['personality'], sot.personality)
self.assertEqual(EXAMPLE['block_device_mapping_v2'],
sot.block_device_mapping)
self.assertEqual(EXAMPLE['os:scheduler_hints'], sot.scheduler_hints)
self.assertEqual(EXAMPLE['user_data'], sot.user_data)
def test_detail(self):
sot = server.ServerDetail()
@ -86,22 +126,15 @@ class TestServer(testtools.TestCase):
self.assertEqual('/servers/detail', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertFalse(sot.allow_create)
self.assertFalse(sot.allow_retrieve)
self.assertFalse(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertTrue(sot.allow_list)
def test_create_body(self):
params = copy.deepcopy(EXAMPLE)
params['scheduler_hints'] = {'group': 'GROUP1_ID'}
body = server.Server._get_create_body(params)
self.assertNotIn('scheduler_hints', body)
self.assertEqual({'group': 'GROUP1_ID'}, body['os:scheduler_hints'])
def test_change_passowrd(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
self.assertEqual(self.resp.body, sot.change_password(self.sess, 'a'))
self.assertIsNone(sot.change_password(self.sess, 'a'))
url = 'servers/IDENTIFIER/action'
body = {"changePassword": {"adminPass": "a"}}
@ -110,9 +143,9 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_reboot(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
self.assertEqual(self.resp.body, sot.reboot(self.sess, 'HARD'))
self.assertIsNone(sot.reboot(self.sess, 'HARD'))
url = 'servers/IDENTIFIER/action'
body = {"reboot": {"type": "HARD"}}
@ -121,21 +154,18 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_rebuild(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
# Let the translate pass through, that portion is tested elsewhere
sot._translate_response = lambda arg: arg
self.assertEqual(
self.resp.body,
sot.rebuild(
self.sess,
name='noo',
image_href='http://image/1',
admin_password='seekr3t',
access_ipv4="12.34.56.78",
access_ipv6="fe80::100",
metadata={"meta var": "meta val"},
personality=[{"path": "/etc/motd", "contents": "foo"}],
)
)
result = sot.rebuild(self.sess, name='noo', admin_password='seekr3t',
image='http://image/1', access_ipv4="12.34.56.78",
access_ipv6="fe80::100",
metadata={"meta var": "meta val"},
personality=[{"path": "/etc/motd",
"contents": "foo"}])
self.assertIsInstance(result, server.Server)
url = 'servers/IDENTIFIER/action'
body = {
@ -147,6 +177,7 @@ class TestServer(testtools.TestCase):
"accessIPv6": "fe80::100",
"metadata": {"meta var": "meta val"},
"personality": [{"path": "/etc/motd", "contents": "foo"}],
"preserve_ephemeral": False
}
}
headers = {'Accept': ''}
@ -154,17 +185,15 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_rebuild_minimal(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
# Let the translate pass through, that portion is tested elsewhere
sot._translate_response = lambda arg: arg
self.assertEqual(
self.resp.body,
sot.rebuild(
self.sess,
name='nootoo',
image_href='http://image/2',
admin_password='seekr3two',
)
)
result = sot.rebuild(self.sess, name='nootoo',
admin_password='seekr3two',
image='http://image/2')
self.assertIsInstance(result, server.Server)
url = 'servers/IDENTIFIER/action'
body = {
@ -172,6 +201,7 @@ class TestServer(testtools.TestCase):
"name": "nootoo",
"imageRef": "http://image/2",
"adminPass": "seekr3two",
"preserve_ephemeral": False
}
}
headers = {'Accept': ''}
@ -179,9 +209,9 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_resize(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
self.assertEqual(self.resp.body, sot.resize(self.sess, '2'))
self.assertIsNone(sot.resize(self.sess, '2'))
url = 'servers/IDENTIFIER/action'
body = {"resize": {"flavorRef": "2"}}
@ -190,9 +220,9 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_confirm_resize(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
self.assertEqual(self.resp.body, sot.confirm_resize(self.sess))
self.assertIsNone(sot.confirm_resize(self.sess))
url = 'servers/IDENTIFIER/action'
body = {"confirmResize": None}
@ -201,9 +231,9 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_revert_resize(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
self.assertEqual(self.resp.body, sot.revert_resize(self.sess))
self.assertIsNone(sot.revert_resize(self.sess))
url = 'servers/IDENTIFIER/action'
body = {"revertResize": None}
@ -212,14 +242,11 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_create_image(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
name = 'noo'
metadata = {'nu': 'image', 'created': 'today'}
self.assertEqual(
self.resp.body,
sot.create_image(self.sess, name, metadata)
)
self.assertIsNone(sot.create_image(self.sess, name, metadata))
url = 'servers/IDENTIFIER/action'
body = {"createImage": {'name': name, 'metadata': metadata}}
@ -228,13 +255,10 @@ class TestServer(testtools.TestCase):
url, endpoint_filter=sot.service, json=body, headers=headers)
def test_create_image_minimal(self):
sot = server.Server(EXAMPLE)
sot = server.Server(**EXAMPLE)
name = 'noo'
self.assertEqual(
self.resp.body,
sot.create_image(self.sess, name)
)
self.assertIsNone(self.resp.body, sot.create_image(self.sess, name))
url = 'servers/IDENTIFIER/action'
body = {"createImage": {'name': name}}
@ -242,38 +266,24 @@ class TestServer(testtools.TestCase):
self.sess.post.assert_called_with(
url, endpoint_filter=dict(sot.service), json=body, headers=headers)
def test_get_ips(self):
name = "jenkins"
fixed = {
"OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:f9:58:b4",
"version": 4,
"addr": "10.3.3.8",
"OS-EXT-IPS:type": "fixed",
}
float1 = {
"OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:f9:58:b4",
"version": 4,
"addr": "15.125.3.1",
"OS-EXT-IPS:type": "floating",
}
float2 = {
"OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:f9:58:b4",
"version": 4,
"addr": "15.125.3.2",
"OS-EXT-IPS:type": "floating",
}
def test_add_security_group(self):
sot = server.Server(**EXAMPLE)
addresses = {name: [fixed]}
attrs = {'id': IDENTIFIER, 'name': name, 'addresses': addresses}
sot = server.Server(attrs=attrs)
self.assertEqual([], sot.get_floating_ips())
self.assertIsNone(sot.add_security_group(self.sess, "group"))
addresses = {name: [fixed, float1, float2]}
attrs = {'id': IDENTIFIER, 'name': name, 'addresses': addresses}
sot = server.Server(attrs=attrs)
self.assertEqual(["15.125.3.1", "15.125.3.2"], sot.get_floating_ips())
url = 'servers/IDENTIFIER/action'
body = {"addSecurityGroup": {"name": "group"}}
headers = {'Accept': ''}
self.sess.post.assert_called_with(
url, endpoint_filter=sot.service, json=body, headers=headers)
addresses = {name: [float1, fixed]}
attrs = {'id': IDENTIFIER, 'name': name, 'addresses': addresses}
sot = server.Server(attrs=attrs)
self.assertEqual(["15.125.3.1"], sot.get_floating_ips())
def test_remove_security_group(self):
sot = server.Server(**EXAMPLE)
self.assertIsNone(sot.remove_security_group(self.sess, "group"))
url = 'servers/IDENTIFIER/action'
body = {"removeSecurityGroup": {"name": "group"}}
headers = {'Accept': ''}
self.sess.post.assert_called_with(
url, endpoint_filter=sot.service, json=body, headers=headers)

7
openstack/tests/unit/compute/v2/test_server_group.py
View File

@ -32,13 +32,16 @@ class TestServerGroup(testtools.TestCase):
self.assertEqual('/os-server-groups', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertTrue(sot.allow_create)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertTrue(sot.allow_delete)
self.assertTrue(sot.allow_list)
self.assertDictEqual({"all_projects": "all_projects"},
sot._query_mapping._mapping)
def test_make_it(self):
sot = server_group.ServerGroup(EXAMPLE)
sot = server_group.ServerGroup(**EXAMPLE)
self.assertEqual(EXAMPLE['id'], sot.id)
self.assertEqual(EXAMPLE['name'], sot.name)
self.assertEqual(EXAMPLE['members'], sot.member_ids)

5
openstack/tests/unit/compute/v2/test_server_interface.py
View File

@ -39,15 +39,14 @@ class TestServerInterface(testtools.TestCase):
self.assertEqual('/servers/%(server_id)s/os-interface', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertTrue(sot.allow_create)
self.assertTrue(sot.allow_retrieve)
self.assertTrue(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertTrue(sot.allow_delete)
self.assertTrue(sot.allow_list)
def test_make_it(self):
sot = server_interface.ServerInterface(EXAMPLE)
sot = server_interface.ServerInterface(**EXAMPLE)
self.assertEqual(EXAMPLE['fixed_ips'], sot.fixed_ips)
self.assertEqual(EXAMPLE['port_id'], sot.id)
self.assertEqual(EXAMPLE['mac_addr'], sot.mac_addr)
self.assertEqual(EXAMPLE['net_id'], sot.net_id)
self.assertEqual(EXAMPLE['port_id'], sot.port_id)

125
openstack/tests/unit/compute/v2/test_server_ip.py
View File

@ -19,97 +19,82 @@ IDENTIFIER = 'IDENTIFIER'
EXAMPLE = {
'addr': '1',
'network_label': '2',
'server_id': '3',
'version': '4',
}
BODY = {
"addresses": {
"public": [
{
"version": 4,
"addr": "67.23.10.132"
},
{
"version": 6,
"addr": "::babe:67.23.10.132"
},
{
"version": 4,
"addr": "67.23.10.131"
},
{
"version": 6,
"addr": "::babe:4317:0A83"
}
],
"private": [
{
"version": 4,
"addr": "10.176.42.16"
},
{
"version": 6,
"addr": "::babe:10.176.42.16"
}
]
}
}
class TestServerIP(testtools.TestCase):
def test_basic(self):
sot = server_ip.ServerIP()
self.assertEqual('server_ip', sot.resource_key)
self.assertEqual('server_ips', sot.resources_key)
self.assertEqual('addresses', sot.resources_key)
self.assertEqual('/servers/%(server_id)s/ips', sot.base_path)
self.assertEqual('compute', sot.service.service_type)
self.assertFalse(sot.allow_create)
self.assertFalse(sot.allow_retrieve)
self.assertFalse(sot.allow_get)
self.assertFalse(sot.allow_update)
self.assertFalse(sot.allow_delete)
self.assertTrue(sot.allow_list)
def test_make_it(self):
sot = server_ip.ServerIP(EXAMPLE)
self.assertEqual(EXAMPLE['addr'], sot.id)
self.assertEqual(EXAMPLE['addr'], sot.addr)
sot = server_ip.ServerIP(**EXAMPLE)
self.assertEqual(EXAMPLE['addr'], sot.address)
self.assertEqual(EXAMPLE['network_label'], sot.network_label)
self.assertEqual(EXAMPLE['server_id'], sot.server_id)
self.assertEqual(EXAMPLE['version'], sot.version)
def test_list(self):
sess = mock.Mock()
resp = mock.Mock()
resp.json = mock.Mock(return_value=BODY)
sess.get = mock.Mock(return_value=resp)
path_args = {'server_id': IDENTIFIER}
sess.get.return_value = resp
resp.json.return_value = {
"addresses": {"label1": [{"version": 1, "addr": "a1"},
{"version": 2, "addr": "a2"}],
"label2": [{"version": 3, "addr": "a3"},
{"version": 4, "addr": "a4"}]}}
caps = server_ip.ServerIP.list(sess, path_args=path_args)
ips = list(server_ip.ServerIP.list(sess, server_id=IDENTIFIER))
caps = sorted(caps, key=lambda cap: cap.id)
self.assertEqual(6, len(caps))
self.assertEqual('10.176.42.16', caps[0].addr)
self.assertEqual('private', caps[0].network_label)
self.assertEqual(IDENTIFIER, caps[0].server_id)
self.assertEqual(4, caps[0].version)
self.assertEqual('67.23.10.131', caps[1].addr)
self.assertEqual('public', caps[1].network_label)
self.assertEqual(IDENTIFIER, caps[1].server_id)
self.assertEqual(4, caps[1].version)
self.assertEqual('67.23.10.132', caps[2].addr)
self.assertEqual('public', caps[2].network_label)
self.assertEqual(IDENTIFIER, caps[2].server_id)
self.assertEqual(4, caps[2].version)
self.assertEqual('::babe:10.176.42.16', caps[3].addr)
self.assertEqual('private', caps[3].network_label)
self.assertEqual(IDENTIFIER, caps[3].server_id)
self.assertEqual(6, caps[3].version)
self.assertEqual('::babe:4317:0A83', caps[4].addr)
self.assertEqual('public', caps[4].network_label)
self.assertEqual(IDENTIFIER, caps[4].server_id)
self.assertEqual(6, caps[4].version)
self.assertEqual('::babe:67.23.10.132', caps[5].addr)
self.assertEqual('public', caps[5].network_label)
self.assertEqual(IDENTIFIER, caps[5].server_id)
self.assertEqual(6, caps[5].version)
self.assertEqual(4, len(ips))
ips = sorted(ips, key=lambda ip: ip.version)
self.assertEqual(type(ips[0]), server_ip.ServerIP)
self.assertEqual(ips[0].network_label, "label1")
self.assertEqual(ips[0].address, "a1")
self.assertEqual(ips[0].version, 1)
self.assertEqual(type(ips[1]), server_ip.ServerIP)
self.assertEqual(ips[1].network_label, "label1")
self.assertEqual(ips[1].address, "a2")
self.assertEqual(ips[1].version, 2)
self.assertEqual(type(ips[2]), server_ip.ServerIP)
self.assertEqual(ips[2].network_label, "label2")
self.assertEqual(ips[2].address, "a3")
self.assertEqual(ips[2].version, 3)
self.assertEqual(type(ips[3]), server_ip.ServerIP)
self.assertEqual(ips[3].network_label, "label2")
self.assertEqual(ips[3].address, "a4")
self.assertEqual(ips[3].version, 4)
def test_list_network_label(self):
label = "label1"
sess = mock.Mock()
resp = mock.Mock()
sess.get.return_value = resp
resp.json.return_value = {label: [{"version": 1,
"addr": "a1"},
{"version": 2,
"addr": "a2"}]}
ips = list(server_ip.ServerIP.list(sess, server_id=IDENTIFIER,
network_label=label))
self.assertEqual(2, len(ips))
ips = sorted(ips, key=lambda ip: ip.version)
self.assertEqual(type(ips[0]), server_ip.ServerIP)
self.assertEqual(ips[0].network_label, label)
self.assertEqual(ips[0].address, "a1")
self.assertEqual(ips[0].version, 1)
self.assertEqual(type(ips[1]), server_ip.ServerIP)
self.assertEqual(ips[1].network_label, label)
self.assertEqual(ips[1].address, "a2")
self.assertEqual(ips[1].version, 2)