ironic_tempest_plugin/services/baremetal/base.py

#    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

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