TeamCityAction

Complexity

Total Complexity

14

Size/Duplication

Total Lines	96
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	_api_post()	0	18	2
A	__init__()	0	8	1
B	_get_baseurl()	0	28	6
A	_api_get()	0	16	2
A	_get_headers()	0	17	3

from st2actions.runners.pythonrunner import Action
import base64

import requests


class TeamCityAction(Action):
    """A simple representation of a TeamCity action.

    :param config: A list, configurations needed for communicating with TeamCity
    """
    def __init__(self, config):
        super(TeamCityAction, self).__init__(config)

        self._version = self.config['version']
        self._username = self.config['username']
        self._password = self.config['password']
        self._url = self._get_baseurl(self.config['url'])
        self._headers = self._get_headers()

    def _get_baseurl(self, url):
        """Configure base url used for all TeamCity api requests

        Args:
            url: The domain name including http/https

        Returns:
            string: URL used for TeamCity api requests
        """
        if not self.config['url']:
            raise ValueError('TeamCity url config must be defined.')

        baseuri = ""

        if url[-1] == '/':
            baseurl = url.rstrip('/')
        else:
            baseurl = url

        if self._username and self._password:
            baseuri = 'httpAuth'

        if self._version:
            baseurl += '/' + baseuri + '/app/rest' + self._version
        else:
            baseurl += '/' + baseuri + '/app/rest'

        return baseurl

    def _get_headers(self):
        """Return headers used in all requests made to the TeamCity api.

        If the a username and password has been configured, then a
        base64 encoding is created for basic authentication.
        """
        headers = {
            'Accept': 'application/json'
        }

        if self._username and self._password:
            auth_header = base64.b64encode('{}:{}'.format(self._username,
                                                          self._password))
            headers['Authorization'] = 'Basic {}'.format(auth_header)
            headers['User-Agent'] = 'Mozilla/5.0'

        return headers

    def _api_get(self, endpoint, params=None):
        """Make get request to the TeamCity api.

         Args:
             endpoint: A string, endpoint uri request is being made to.
             params: A list, parameters to be added as query string members in request
        Returns:
            string: Json string response
        """
        if endpoint[0] == '/':
            endpoint = endpoint.lstrip('/')

        url = '/'.join([self._url, endpoint])
        r = requests.get(url, params, headers=self._headers)

There seem to be too many positional arguments for this function call.

        r.raise_for_status()
        return r.json()

    def _api_post(self, endpoint, data):
        """Make post request to the TeamCity api.

         Args:
             endpoint: A string, endpoint uri request is being made to.
             data: Json data to be added to the body of the request
        Returns:
            string: Json string response
        """
        if endpoint[0] == '/':
            endpoint = endpoint.lstrip('/')

        self._headers['Content-Type'] = 'application/xml'

        url = '/'.join([self._url, endpoint])
        r = requests.post(url, data=data, headers=self._headers)
        r.raise_for_status()
        return r.json()