[Kimchi-devel] [PATCH V2][Wok 04/12] FVT: Common classes/methods for API calls as per config file configuration.
archus at linux.vnet.ibm.com
archus at linux.vnet.ibm.com
Mon May 30 07:10:23 UTC 2016
From: Archana Singh <archus at linux.vnet.ibm.com>
Wrapper classes/methods for making API calls by reading
the config file configuration.
Signed-off-by: Archana Singh <archus at linux.vnet.ibm.com>
---
tests/fvt/restapilib.py | 738 ++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 738 insertions(+)
create mode 100644 tests/fvt/restapilib.py
diff --git a/tests/fvt/restapilib.py b/tests/fvt/restapilib.py
new file mode 100644
index 0000000..19634f1
--- /dev/null
+++ b/tests/fvt/restapilib.py
@@ -0,0 +1,738 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#
+# Project Wok
+#
+# Copyright IBM Corp, 2016
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301USA
+
+import ConfigParser
+import jsonschema
+import json
+import requests
+import logging
+import os
+
+requests.packages.urllib3.disable_warnings()
+
+__all__ = ['APIError', 'APIRequestError', 'APISession', 'SessionParameters']
+
+DEFAULT_CONF = os.path.dirname(os.path.abspath(__file__)) + os.sep + 'config'
+
+_HTTP_SUCCESS_STATUS_RANGE = range(200, 299) # All of the 2XX status codes
+
+
+class APIError(Exception):
+ """Base class for exceptions raised by this module."""
+
+ pass
+
+
+class APISession(object):
+ """Represents an API session with the host Web Services API"""
+
+ def __init__(self, conffile=DEFAULT_CONF):
+ """ Creates an APISession object for use with the specified Host.
+ """
+ self._session = None
+ self._base_uri = None
+ self.username = None
+ self.passwd = None
+ self.host = None
+ self.port = None
+ # Holds the session from requests if self._session is None.
+ self.session()
+ self.sessionparams = SessionParameters(conffile=conffile)
+ self.logging = self.sessionparams.getLogger()
+ # base URI of connected machine
+ self._base_uri = self.create_base_uri()
+
+ def session(self):
+ """
+ Create the session object for API Request if no existing session.
+ """
+ if self._session is not None:
+ raise APIError('Already have a session')
+
+ self._session = requests.Session()
+ return self._session
+
+ def auth(self):
+ """
+ Set authentication to API session.
+ """
+
+ if self._session is None:
+ self.session()
+
+ self.username, self.passwd = self.sessionparams.getCredentials()
+
+ if self.username is None:
+ raise APIError('Username is None')
+ if self.passwd is None:
+ raise APIError('password is None')
+ # Set auth at session level
+ self._session.auth = (self.username, self.passwd)
+
+ # set password expiration currently it default value
+ # self._session.expires = '1432091741'
+ # For now keeping it False to disable SSL verification
+ self._session.verify = False
+ return
+
+ def create_base_uri(self):
+ """
+ Create the base URI using host and port.
+ if no host is provided _base_uri is None.
+ if port is None then _base_uri is just host
+ if both is provided then append port to host.
+ """
+
+ if self._base_uri is not None:
+ return self._base_uri
+
+ self.host, self.port = self.sessionparams.getUrlParams()
+ if self.host is None:
+ return None
+ if self.port is None:
+ return self.host
+
+ self._base_uri = 'https://' + self.host + ':' + self.port
+ return self._base_uri
+
+ def end_session(self):
+ """Ends an API session"""
+
+ if self._session is None:
+ return None # Silently ignore
+ # The session to be terminated by setting None in session object we
+ # have.
+ self._session = None
+ return
+
+ def request(
+ self,
+ method,
+ uri,
+ body=None,
+ expected_status_values=None,
+ headers=None, ):
+ """
+ Issue an WS API request and return the response body, if any.
+
+ \param method
+ The HTTP method to issue (eg. GET, PUT, POST, DELETE).
+
+ \param uri
+ The URI path and query parameter string for the request.
+
+ \param body
+ The request body that is input to the request, as a Unicode
+ or String, or None if there is no input body.
+
+ \param expected_status_values
+ The HTTP status code that is expected on completion of the request.
+ If this parameter is specified, the function raises an exception
+ if the HTTP status from the request was not exactly as specified.
+ Otherwise, it raises an error if the HTTP status is not 2XX.
+
+ \param headers
+ Request headers for this request, in the form of a Python Dict.
+ Optional. This function automatically augments these headers
+ with the headers needed to specify the API session. Note that if
+ input headers are provided, the supplied Dict object will be
+ modified by this function.
+ """
+
+ # Start with the headers supplied by caller, if any
+
+ if headers is not None:
+ hdrs = headers # Use caller's dict, not a copy
+ else:
+ hdrs = dict()
+
+ # If no session then throw error.
+
+ if self._session is None:
+ raise APIError('You have no session')
+
+ resp = self._session.request(method, uri, data=body, headers=hdrs)
+
+ # If an expected status was specified, check that the status exactly
+ # matches wbat was expected, otherwise check that the status is one of
+ # the success Statuses.
+
+ if expected_status_values is not None:
+ raise_exc = resp.status_code not in expected_status_values
+ else:
+ raise_exc = resp.status_code not in _HTTP_SUCCESS_STATUS_RANGE
+
+ # Raise an exceptoin if the result is not as intended.
+
+ if raise_exc:
+
+ # If the request fails in some way WEb Services API response
+ # will usually include a standard error response body
+ # in JSON format that includes a more detailed reason
+ # code (and message) for the failure. It provides this data
+ # in JSON format even if the request would return some other
+ # format if the request had been successful.
+ # So if the request has failed, grab that additional info
+ # for use in raising exceptions below.
+
+ failure_reason = 0
+ failure_code = None
+ # if response.status_code not in _HTTP_SUCCESS_STATUS_RANGE:
+
+ # The API provides the JSON error response in all usual error
+ # cases. But for certain less common errors this does not occur
+ # because the error is caught higher in the processing stack.
+ # So try to interpret the response as a JSON response body, but
+ # just silently ignore problems if we can't do this.
+
+ try:
+ error_resp = json.loads(resp.text)
+ failure_reason = error_resp['reason']
+ failure_code = error_resp['code']
+ except (ValueError, KeyError):
+ pass
+
+ raise APIRequestError(resp.status_code, failure_reason,
+ failure_code)
+
+ # Return the response
+ return resp
+
+ def request_octet(
+ self,
+ method,
+ uri,
+ body=None,
+ expected_status_values=None,
+ headers=None,
+ ):
+
+ # Start with the request headers the caller supplied if any,
+ # otherwise start with an empty set of headers.
+
+ if headers is not None:
+ hdrs = headers # Using caller's dict, not a copy
+ else:
+ hdrs = dict()
+
+ # If a request body was specified, convert it from its assumed Dict
+ # or List form into JSON using the Python JSON library. Also, add in
+ # the required Content-Type HTTP message header to let the API know
+ # that we are supplying input in JSON form. (A Content-Length header
+ # is also needed to specify the body length, but for not not set.)
+
+ body_json = body
+ if body is not None:
+ body_json = json.dumps(body)
+
+ # Supply an HTTP Accepts header indicating we accept/expect that any
+ # response body be JSON. The Accepts header is optional, and the API
+ # will defautl to supplying JSON for any operation defined in API V1.1
+ # to return JSON (even if in the future support for other formats would
+ # be added). But its a good idea to specify this header as a safeguard
+ # in case this function were called for an URI that is not capable of
+ # returning JSON, as this function as currently coded would choke on
+ # non-JSON responses.
+
+ hdrs['Accept'] = 'application/json'
+
+ # Now issue the request, and retrieve the response object and the
+ # response body if provided.
+
+ resp = self.request(
+ method,
+ uri,
+ body=body_json,
+ expected_status_values=expected_status_values,
+ headers=hdrs,
+ )
+
+ return resp
+
+ def request_json(
+ self,
+ method,
+ uri,
+ body=None,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """
+ Issue an WS API request that is defined to take JSON input and
+ produce JSON output. (Nearly all WS API requests are like this.)
+
+ Input and output bodies are specified as Python Dict or List objects,
+ which are converted to/from JSON by this function.
+
+ \param method
+ The HTTP method to issue (eg. GET, PUT, POST, DELETE).
+
+ \param uri
+ The URI path and query parameter string for the request.
+
+ \param body
+ The request body that is input to the request, in the form of a
+ Python Dict or List object. This object is automatically converted
+ to corresponding JSON by this function. Optional.
+
+ \param expected_status_values
+ The HTTP status code that is expected on completion of the request.
+ See corresponding parameter on request() method for more details.
+
+ \param headers
+ Request headers for this request, in the form of a Python Dict.
+ Optional. This function automatically augments these headers
+ with the headers needed to specify the JSON content type for input
+ and output, and to specify the API session. Note that if input
+ headers are provided, the supplied Dict object will be modified
+ by this function.
+
+ """
+
+ # Start with the request headers the caller supplied if any,
+ # otherwise start with an empty set of headers.
+
+ if headers is not None:
+ hdrs = headers # Using caller's dict, not a copy
+ else:
+ hdrs = dict()
+
+ # If a request body was specified, convert it from its assumed Dict
+ # or List form into JSON using the Python JSON library. Also, add in
+ # the required Content-Type HTTP message header to let the API know
+ # that we are supplying input in JSON form. (A Content-Length header
+ # is also needed to specify the body length, but for now have not set.)
+
+ body_json = None
+ if body is not None:
+ body_json = json.dumps(body)
+ hdrs['Content-Type'] = 'application/json'
+
+ # Supply an HTTP Accepts header indicating we accept/expect that any
+ # response body be JSON. The Accepts header is optional, and the API
+ # will default to supplying JSON for any operation defined in API V1.1
+ # to return JSON (even if in the future support for other formats would
+ # be added). But its a good idea to specify this header as a safeguard
+ # in case this function were called for an URI that is not capable of
+ # returning JSON, as this function as currently coded would choke on
+ # non-JSON responses.
+
+ hdrs['Accept'] = 'application/json'
+
+ # Now issue the request, and retrieve the response object and the
+ # response body if provided.
+
+ response_body = self.request(
+ method,
+ uri,
+ body=body_json,
+ expected_status_values=expected_status_values,
+ headers=hdrs,
+ )
+
+ # If a response body was returned, we presume it is JSON and
+ # convert it into a Python dict we will return.
+
+ resp_json = None
+ if response_body is not None:
+ try:
+ resp_json = response_body.json()
+ except ValueError, err:
+ print 'response_body %s' % response_body
+ print 'response_body headers %s' % response_body.headers
+ raise APIError('Response body expected to be JSON but is'
+ + 'not valid %s', err)
+ return resp_json
+
+ def request_get_json(
+ self,
+ uri,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a GET request that returns JSON."""
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ return self.request_json('GET', uri,
+ expected_status_values=expected_status_values,
+ headers=headers)
+
+ def request_put_json(
+ self,
+ uri,
+ body=None,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a GET request that returns JSON."""
+
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ return self.request_json(
+ 'PUT',
+ uri,
+ body=body,
+ expected_status_values=expected_status_values,
+ headers=headers,
+ )
+
+ def request_post_json(
+ self,
+ uri,
+ body=None,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a POST request that
+ provides/returns JSON."""
+
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ return self.request_json(
+ 'POST',
+ uri,
+ body=body,
+ expected_status_values=expected_status_values,
+ headers=headers,
+ )
+
+ def request_delete_json(
+ self,
+ uri,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a DELETE request that returns JSON."""
+
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ return self.request_json(
+ 'DELETE',
+ uri,
+ expected_status_values=expected_status_values,
+ headers=headers,
+ )
+
+ def request_get(
+ self,
+ uri,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a GET request with header['Accept']
+ as application/json that returns the response.
+ Does not convert the response to JSON
+ """
+
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ if headers is None:
+ headers = dict()
+ headers['Accept'] = 'application/json'
+
+ return self.request('GET', uri,
+ expected_status_values=expected_status_values,
+ headers=headers)
+
+ def request_put(
+ self,
+ uri,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a PUT request with header['Accept']
+ as application/json that returns the response.
+ Does not convert the response to JSON
+ """
+
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ if headers is None:
+ headers = dict()
+ headers['Accept'] = 'application/json'
+
+ return self.request(
+ 'PUT',
+ uri,
+ expected_status_values=expected_status_values,
+ headers=headers,
+ )
+
+ def request_post(
+ self,
+ uri,
+ body=None,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a POST request with header['Accept']
+ as application/json that returns the response.
+ Does not convert the response to JSON
+ """
+
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ # If a request body was specified, convert it from its assumed Dict
+ # or List form into JSON using the Python JSON library. Also, add in
+ # the required Content-Type HTTP message header to let the API know
+ # that we are supplying input in JSON form. (A Content-Length header
+ # is also needed to specify the body length, but for now have not set.)
+
+ body_json = None
+ if body is not None:
+ body_json = json.dumps(body)
+
+ if headers is None:
+ headers = dict()
+ headers['Content-Type'] = 'application/json'
+ headers['Accept'] = 'application/json'
+
+ return self.request(
+ 'POST',
+ uri,
+ body=body_json,
+ expected_status_values=expected_status_values,
+ headers=headers,
+ )
+
+ def request_delete(
+ self,
+ uri,
+ expected_status_values=None,
+ headers=None,
+ ):
+ """Convenience function to do a DELETE request with header['Accept']
+ as application/json that returns the response.
+ Does not convert the response to JSON
+ """
+
+ if self._base_uri is not None:
+ uri = self._base_uri + uri
+
+ if headers is None:
+ headers = dict()
+ headers['Accept'] = 'application/json'
+
+ return self.request(
+ 'DELETE',
+ uri,
+ expected_status_values=expected_status_values,
+ headers=headers,
+ )
+
+
+class APIRequestError(APIError):
+ """
+ Raised when an API request ends in error or not as expected.
+
+ Attributes:
+ \param status
+ HTTP status code from the request
+
+ \param reason
+ API reason code from the request
+
+ \param message
+ API diagnostic message from the request
+
+ \param stack
+ Internal diagnostic info for selected Status 500 errors
+ """
+
+ def __init__(
+ self,
+ status,
+ reason=None,
+ code=None,
+ ):
+ self.status = status
+ self.reason = reason
+ self.code = code
+
+ def __str__(self):
+ if self.code is not None:
+ s = 'Request ended with status %s-%s (%s)' \
+ % (self.status, self.reason, self.code)
+ else:
+ s = 'Request ended with status %s-%s' % (self.status,
+ self.reason)
+ return s
+
+
+class BadJSONResponse(APIError):
+ """exceptions raised for Bad Json Response."""
+
+ def __init__(
+ self,
+ error,
+ ):
+ self.error = error
+
+ def __str__(self):
+ s = 'Bad response: %s' % self.error
+ return s
+
+
+class InvalidInput(APIError):
+ """exceptions raised for Invalid Input."""
+
+ def __init__(
+ self,
+ error,
+ ):
+ self.error = error
+
+ def __str__(self):
+ s = 'Bad response: %s' % self.error
+ return s
+
+
+class SessionParameters(object):
+ """
+ Represents functionalities that could help in getting the
+ session information.
+ """
+
+ def __init__(
+ self,
+ conffile=DEFAULT_CONF):
+ """
+ Attributes:
+ \param conffile
+ config file which contains all configuration
+ information with sections
+ """
+ self.conffile = conffile
+ self.username = None
+ self.passwd = None
+ self.host = '127.0.0.1'
+ self.port = '8001'
+ self.params = 'config'
+ self.logfile = 'wok-api-test-suite.log'
+ self.session_section = 'Session'
+ self.logging = logging
+ self.loglevel = 'ERROR'
+ self.LEVELS = {'INFO': self.logging.INFO,
+ 'DEBUG': self.logging.DEBUG,
+ 'WARNING': self.logging.WARNING,
+ 'ERROR': self.logging.ERROR,
+ 'CRITICAL': self.logging.CRITICAL}
+ if self.conffile is None:
+ print 'Configuration file required %s' \
+ % self.conffile
+ else:
+ print 'Reading configuration file %s' % self.conffile
+ self.params = ConfigParser.ConfigParser()
+ print self.params.read(self.conffile)
+ if self.params.has_section(self.session_section):
+ if self.params.has_option(self.session_section, 'logfile'):
+ self.logfile = self.params.get(
+ self.session_section, 'logfile')
+
+ if self.params.has_option(self.session_section, 'loglevel'):
+ self.loglevel = self.params.get(
+ self.session_section, 'loglevel')
+ else:
+ print "Section %s is not available in the config file " \
+ % self.session_section
+
+ self.logging.basicConfig(format='%(asctime)s %(levelname)s: \
+ %(message)s',
+ filename=self.logfile,
+ level=self.LEVELS[self.loglevel])
+
+ def getCredentials(self):
+ """
+ Returns user and password details for a session obtained from config
+ file
+ """
+ if self.params.has_section(self.session_section):
+ if self.params.has_option(self.session_section, 'user'):
+ self.username = self.params.get(self.session_section, 'user')
+
+ if self.params.has_option(self.session_section, 'passwd'):
+ self.passwd = self.params.get(self.session_section, 'passwd')
+ else:
+ print "Section %s is not available in the config file " \
+ % self.session_section
+ raise APIError('Session section not present in config')
+
+ return self.username, self.passwd
+
+ def getUrlParams(self):
+ """
+ Returns host and port details for a session obtained from config file
+ """
+ if self.params.has_section(self.session_section):
+ if self.params.has_option(self.session_section, 'host'):
+ self.host = self.params.get(self.session_section, 'host')
+
+ if self.params.has_option(self.session_section, 'port'):
+ self.port = self.params.get(self.session_section, 'port')
+ else:
+ print "Section %s is not available in the config file " \
+ % self.session_section
+
+ return self.host, self.port
+
+ def getLogger(self):
+ return self.logging
+
+
+class Validator(object):
+ """
+ validator class having different validate methods
+ """
+
+ def __init__(self, logger=None):
+ """
+ :param logger:
+ :return:
+ """
+ if logger is not None:
+ self.logging = logger
+
+ def validate_json(self, jsn, schma):
+ """
+ validates json against schema using jsonschema.validate library
+
+ :param jsn: JSON to validate
+ :param schma: Schema against which JSON should get validated
+ :return:
+ """
+ if self.logging is not None:
+ self.logging.info('--> validate_json()')
+ self.logging.debug(
+ 'validate_json(): jsn:%s' % jsn)
+ self.logging.debug(
+ 'validate_json(): schma:%s' % schma)
+ jsonschema.validate(jsn, schma)
+ if self.logging is not None:
+ self.logging.debug(
+ 'validate_json(): jsn is valid')
+ self.logging.info('<-- validate_json()')
--
2.5.0
More information about the Kimchi-devel
mailing list