bugzilla.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

# TODO:
# 1. Load REST URL from system wide config
# 2. New content_type for conduit attachments?
# 3. Add comment_tags for conduit attachments?

"""Interface to a Bugzilla system."""


from urllib.parse import quote
import json
import logging
import requests


class Bugzilla(object):
    """
    Interface to a Bugzilla system.
    """

    def __init__(self, rest_url=None):
        self.rest_url = rest_url
        self.session = requests.Session()
        self.logger = logging.getLogger(__name__)

    def call(self, method, path, data=None):
        """Perform REST API call and decode JSON"""
        headers = {
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }

        if method == 'GET':
            response = self.session.get(self.rest_url + path, headers=headers)

        if method == 'POST':
            response = self.session.post(self.rest_url + path, json=data,
                                         headers=headers)

        print(response.content)
        data = json.loads(response.content.decode('utf-8'))

        if isinstance(data, dict) and 'error' in data and data['error']:
            raise BugzillaError(data['code'], data['message'])

        return data

    def is_bug_confidential(self, bug_id):
        """Check if bug is confidential"""
        self.logger.info('Checking if bug %d is confidential.', bug_id)
        try:
            self.call('GET', '/bug/' + quote(str(bug_id)))
        except BugzillaError as error:
            if error.fault_code == 102:
                return True
        except:
            raise BugzillaError(error.fault_code, error.fault_string)

        return False

    def valid_api_key(self, username, api_key):
        """Check if API key is valid for specific username"""
        self.logger.info('Checking valid API key for %s.', username)
        try:
            self.call('GET', '/valid_login?login=' + quote(username) +
                      '&api_key=' + quote(api_key))
        except BugzillaError as error:
            if error.fault_code == 306:
                return False
        except:
            raise BugzillaError(error.fault_code, error.fault_string)

        return True

    def create_attachment(self, bug_id, attach_data, api_key=None):
        """Create the attachment using the provided flags.

        The `flags` parameter is an array of flags to set/update/clear.  This
        array matches the Bugzilla flag API:
        Setting:
            {
                'id': flag.id
                'name': 'review',
                'status': '?',
                'requestee': reviewer.email
            }
        Clearing:
            {
                'id': flag.id,
                'status': 'X'
            }
        """

        self.logger.info('Posting attachment to bug %d.', bug_id)

        try:
            result = self.call('POST', '/bug/' + quote(str(bug_id)) +
                               '/attachment?api_key=' + quote(str(api_key)),
                               attach_data)
        except BugzillaError as error:
            print('code: %d string: %s' % (error.fault_code, error.fault_string))
            return None

        return list(result['attachments'].keys())[0]


class BugzillaError(Exception):
    """Generic Bugzilla Exception"""
    def __init__(self, msg, code=None):
        super(BugzillaError, self).__init__(msg)
        self.fault_code = code
        self.fault_string = msg