Source code for cup.jenkinslib.internal.requester

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Copyright: [CUP] - See LICENSE for details.
# Authors: liushuxian(liushuxian)
"""
This module provides some requester to access jenkins.
"""

from cup.thirdp import requests
from cup.jenkinslib.internal import exception


class Requester(object):
    """A class which help you access jenkins."""
    VALID_STATUS_CODES = [200]

    def __init__(self, username=None, password=None):
        """initialize Requester object."""
        self.username = username
        self.password = password

    def get(self, url, params=None, headers=None, allow_redirects=True):
        """request url in GET method.

        Returns:
            Response object.
        """
        requests_kwargs = self.__build_request(params=params, headers=headers,
                                               allow_redirects=allow_redirects)
        try:
            return requests.get(url, **requests_kwargs)
        except requests.RequestException as err:
            raise exception.RequestError(url, "GET", err=err)

    def post(self, url, params=None, data=None, files=None, headers=None, allow_redirects=True):
        """request url in POST method.

        Returns:
            Response object.
        """
        requests_kwargs = self.__build_request(params=params, headers=headers,
                                               data=data, files=files,
                                               allow_redirects=allow_redirects)
        try:
            return requests.post(url, **requests_kwargs)
        except requests.RequestException as err:
            raise exception.RequestError(url, "POST", err=err)

    def get_and_confirm_status(self, url, params=None, headers=None, valid=None):
        """request url in GET method, and check status code.

        Returns:
            Response object.
        """
        valid = valid or self.VALID_STATUS_CODES
        response = self.get(url, params, headers)
        if response.status_code not in valid:
            if response.status_code == 405:         # POST required
                raise exception.PostRequired(response.url, "GET", response.status_code)
            else:
                raise exception.InvalidRequestStatus(
                    response.url,
                    method="GET",
                    status=response.status_code,
                    msg="operation failed.",
                    response=response)
        return response

    def post_and_confirm_status(self, url, params=None, data=None, files=None,
                                headers=None, valid=None, allow_redirects=True):
        """request url in POST method, and check status code.

        Returns:
            Response object.
        """
        valid = valid or self.VALID_STATUS_CODES
        if not isinstance(data, (str, dict)):
            raise exception.ParamTypeError(
                "unexpected type of parameter 'data': %s. Expected (str, dict)" % type(data))

        if not headers and not files:
            headers = {'Content-Type': 'application/x-www-form-urlencoded'}

        response = self.post(url, params, data, files, headers, allow_redirects)
        if response.status_code not in valid:
            raise exception.InvalidRequestStatus(
                response.url,
                method="POST",
                status=response.status_code,
                msg="operation failed.",
                response=response)
        return response

    def post_xml_and_confirm_status(self, url, params=None, data=None, valid=None):
        """request url in POST method with text/xml context, and check status code.

        Returns:
            Response object.
        """
        headers = {'Content-Type': 'text/xml'}
        return self.post_and_confirm_status(
            url, params=params, data=data, headers=headers, valid=valid)

    def __build_request(self, params=None, data=None, files=None, headers=None, **kwargs):
        """build kwargs for requests."""
        requests_kwargs = kwargs

        if self.username:
            requests_kwargs["auth"] = (self.username, self.password)

        if headers:
            if not isinstance(headers, dict):
                raise exception.ParamTypeError(
                    "unexpected type of parameter 'headers': %s. Expected (dict)" % type(data))
            requests_kwargs["headers"] = headers

        if params:
            if not isinstance(params, dict):
                raise exception.ParamTypeError(
                    "unexpected type of parameter 'params': %s. Expected (dict)" % type(data))
            requests_kwargs["params"] = params

        if data:
            if not isinstance(data, (str, dict)):
                raise exception.ParamTypeError(
                    "unexpected type of parameter 'data': %s. Expected (str, dict)" % type(data))
            requests_kwargs["data"] = data

        if files:
            requests_kwargs["files"] = files

        return requests_kwargs