openstack/ironic
Code Issues Proposed changes
5ebc60de42
ironic/ironic_tempest_plugin/services/baremetal/base.py
Vasyl Saienko e48375538c Add Ironic standalone tests 
This patch adds the following standalone tests:

  * agent_ipmitool + wholedisk image + bios
  * agent_ipmitool + partitioned image + bios
  * pxe_ipmitool + wholedisk image + bios
  * pxe_ipmitool + partitioned image + bios

Partial-Bug: #1660606

Change-Id: Ic04b0f134e20d9937a610a14d7c4128f45738eeb
2017-03-15 16:54:45 +02:00

261 lines
9.0 KiB
Python
Raw Blame History

# 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 functools
from oslo_serialization import jsonutils as json
from six.moves import http_client
from six.moves.urllib import parse as urllib
from tempest.lib.common import api_version_utils
from tempest.lib.common import rest_client
# NOTE(vsaienko): concurrent tests work because they are launched in
# separate processes so global variables are not shared among them.
BAREMETAL_MICROVERSION = None
def set_baremetal_api_microversion(baremetal_microversion):
global BAREMETAL_MICROVERSION
BAREMETAL_MICROVERSION = baremetal_microversion
def reset_baremetal_api_microversion():
global BAREMETAL_MICROVERSION
BAREMETAL_MICROVERSION = None
def handle_errors(f):
"""A decorator that allows to ignore certain types of errors."""
@functools.wraps(f)
def wrapper(*args, **kwargs):
param_name = 'ignore_errors'
ignored_errors = kwargs.get(param_name, tuple())
if param_name in kwargs:
del kwargs[param_name]
try:
return f(*args, **kwargs)
except ignored_errors:
# Silently ignore errors
pass
return wrapper
class BaremetalClient(rest_client.RestClient):
"""Base Tempest REST client for Ironic API."""
api_microversion_header_name = 'X-OpenStack-Ironic-API-Version'
uri_prefix = ''
def get_headers(self):
headers = super(BaremetalClient, self).get_headers()
if BAREMETAL_MICROVERSION:
headers[self.api_microversion_header_name] = BAREMETAL_MICROVERSION
return headers
def request(self, *args, **kwargs):
resp, resp_body = super(BaremetalClient, self).request(*args, **kwargs)
if (BAREMETAL_MICROVERSION and
BAREMETAL_MICROVERSION != api_version_utils.LATEST_MICROVERSION):
api_version_utils.assert_version_header_matches_request(
self.api_microversion_header_name,
BAREMETAL_MICROVERSION,
resp)
return resp, resp_body
def serialize(self, object_dict):
"""Serialize an Ironic object."""
return json.dumps(object_dict)
def deserialize(self, object_str):
"""Deserialize an Ironic object."""
return json.loads(object_str)
def _get_uri(self, resource_name, uuid=None, permanent=False):
"""Get URI for a specific resource or object.
:param resource_name: The name of the REST resource, e.g., 'nodes'.
:param uuid: The unique identifier of an object in UUID format.
:returns: Relative URI for the resource or object.
"""
prefix = self.uri_prefix if not permanent else ''
return '{pref}/{res}{uuid}'.format(pref=prefix,
res=resource_name,
uuid='/%s' % uuid if uuid else '')
def _make_patch(self, allowed_attributes, **kwargs):
"""Create a JSON patch according to RFC 6902.
:param allowed_attributes: An iterable object that contains a set of
allowed attributes for an object.
:param **kwargs: Attributes and new values for them.
:returns: A JSON path that sets values of the specified attributes to
the new ones.
"""
def get_change(kwargs, path='/'):
for name, value in kwargs.items():
if isinstance(value, dict):
for ch in get_change(value, path + '%s/' % name):
yield ch
else:
if value is None:
yield {'path': path + name,
'op': 'remove'}
else:
yield {'path': path + name,
'value': value,
'op': 'replace'}
patch = [ch for ch in get_change(kwargs)
if ch['path'].lstrip('/') in allowed_attributes]
return patch
def _list_request(self, resource, permanent=False, headers=None,
extra_headers=False, **kwargs):
"""Get the list of objects of the specified type.
:param resource: The name of the REST resource, e.g., 'nodes'.
:param headers: List of headers to use in request.
:param extra_headers: Specify whether to use headers.
:param **kwargs: Parameters for the request.
:returns: A tuple with the server response and deserialized JSON list
of objects
"""
uri = self._get_uri(resource, permanent=permanent)
if kwargs:
uri += "?%s" % urllib.urlencode(kwargs)
resp, body = self.get(uri, headers=headers,
extra_headers=extra_headers)
self.expected_success(http_client.OK, resp.status)
return resp, self.deserialize(body)
def _show_request(self, resource, uuid, permanent=False, **kwargs):
"""Gets a specific object of the specified type.
:param uuid: Unique identifier of the object in UUID format.
:returns: Serialized object as a dictionary.
"""
if 'uri' in kwargs:
uri = kwargs['uri']
else:
uri = self._get_uri(resource, uuid=uuid, permanent=permanent)
resp, body = self.get(uri)
self.expected_success(http_client.OK, resp.status)
return resp, self.deserialize(body)
def _create_request(self, resource, object_dict):
"""Create an object of the specified type.
:param resource: The name of the REST resource, e.g., 'nodes'.
:param object_dict: A Python dict that represents an object of the
specified type.
:returns: A tuple with the server response and the deserialized created
object.
"""
body = self.serialize(object_dict)
uri = self._get_uri(resource)
resp, body = self.post(uri, body=body)
self.expected_success(http_client.CREATED, resp.status)
return resp, self.deserialize(body)
def _create_request_no_response_body(self, resource, object_dict):
"""Create an object of the specified type.
Do not expect any body in the response.
:param resource: The name of the REST resource, e.g., 'nodes'.
:param object_dict: A Python dict that represents an object of the
specified type.
:returns: The server response.
"""
body = self.serialize(object_dict)
uri = self._get_uri(resource)
resp, body = self.post(uri, body=body)
self.expected_success(http_client.NO_CONTENT, resp.status)
return resp
def _delete_request(self, resource, uuid):
"""Delete specified object.
:param resource: The name of the REST resource, e.g., 'nodes'.
:param uuid: The unique identifier of an object in UUID format.
:returns: A tuple with the server response and the response body.
"""
uri = self._get_uri(resource, uuid)
resp, body = self.delete(uri)
self.expected_success(http_client.NO_CONTENT, resp.status)
return resp, body
def _patch_request(self, resource, uuid, patch_object):
"""Update specified object with JSON-patch.
:param resource: The name of the REST resource, e.g., 'nodes'.
:param uuid: The unique identifier of an object in UUID format.
:returns: A tuple with the server response and the serialized patched
object.
"""
uri = self._get_uri(resource, uuid)
patch_body = json.dumps(patch_object)
resp, body = self.patch(uri, body=patch_body)
self.expected_success(http_client.OK, resp.status)
return resp, self.deserialize(body)
@handle_errors
def get_api_description(self):
"""Retrieves all versions of the Ironic API."""
return self._list_request('', permanent=True)
@handle_errors
def get_version_description(self, version='v1'):
"""Retrieves the description of the API.
:param version: The version of the API. Default: 'v1'.
:returns: Serialized description of API resources.
"""
return self._list_request(version, permanent=True)
def _put_request(self, resource, put_object):
"""Update specified object with JSON-patch."""
uri = self._get_uri(resource)
put_body = json.dumps(put_object)
resp, body = self.put(uri, body=put_body)
self.expected_success([http_client.ACCEPTED, http_client.NO_CONTENT],
resp.status)
return resp, body
View Git Blame Copy Permalink