path: root/buildbot_gentoo_ci/steps/bugs.py

# Copyright 2023 Gentoo Authors
# Distributed under the terms of the GNU General Public License v2

# Origins: bugz.cli
# Modifyed by Gentoo Authors.

import re
import os
import xmlrpc

from twisted.internet import defer

from buildbot.process.buildstep import BuildStep
from buildbot.process.results import SUCCESS
from buildbot.process.results import FAILURE
from buildbot.process.results import SKIPPED
from buildbot.plugins import util

from buildbot_gentoo_ci.utils.regex import stripQuotesAndMore, finishTitle

from bugz.cli import check_bugz_token, login, list_bugs
from bugz.cli_argparser import make_arg_parser
from bugz.configfile import load_config
from bugz.settings import Settings
from bugz.exceptions import BugzError
from bugz.log import log_error, log_info

from portage.versions import cpv_getversion, pkgsplit, catpkgsplit

# main
def main_bugz(args):
    ArgParser = make_arg_parser()
    opt = ArgParser.parse_args(args)

    ConfigParser = load_config(getattr(opt, 'config_file', None))

    check_bugz_token()
    settings = Settings(opt, ConfigParser)
    return settings

# search
def search_bugz(args):
    settings = main_bugz(args)
    valid_keys = ['alias', 'assigned_to', 'component', 'creator',
                'limit', 'offset', 'op_sys', 'platform',
                'priority', 'product', 'resolution', 'severity',
                'version', 'whiteboard', 'cc']

    params = {}
    d = vars(settings)
    for key in d:
        if key in valid_keys:
            params[key] = d[key]
    if 'search_statuses' in d:
        if 'all' not in d['search_statuses']:
            params['status'] = d['search_statuses']
    if 'terms' in d:
        params['summary'] = d['terms']

    if not params:
        raise BugzError('Please give search terms or options.')

    log_info('Searching for bugs meeting the following criteria:')
    for key in params:
        log_info('   {0:<20} = {1}'.format(key, params[key]))

    login(settings)

    result = settings.call_bz(settings.bz.Bug.search, params)['bugs']

    if not len(result):
        log_info('No bugs found.')
        return []
    else:
        list_bugs(result, settings)
        return result

# post
def post_bugs(args, params):
    """Post a new bug"""
    settings = main_bugz(args)
    result = settings.call_bz(settings.bz.Bug.create, params)
    log_info('Bug %d submitted' % result['id'])
    return result

# modify
def modify_bugs(args, params):
    """Modify an existing bug (eg. adding a comment or changing resolution.)"""
    settings = main_bugz(args)
    if len(params) < 2:
        raise BugzError('No changes were specified')
    result = settings.call_bz(settings.bz.Bug.update, params)
    return result

def attach_bugs(args, params):
    """ Attach a file to a bug given a filename. """
    settings = main_bugz(args)
    fd = open(params['filename'], 'rb')
    params['data'] = xmlrpc.client.Binary(fd.read())
    fd.close()
    result =  settings.call_bz(settings.bz.Bug.add_attachment, params)
    return result

class GetBugs(BuildStep):
    
    name = 'GetBugs'
    description = 'Running'
    descriptionDone = 'Ran'
    descriptionSuffix = None
    haltOnFailure = True
    flunkOnFailure = True

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    @defer.inlineCallbacks
    def find_match(self, buglist):
        log = yield self.addLog('Bugs')
        yield log.addStdout('Open Bugs\n')
        match = False
        for bug in buglist:
            yield log.addStdout(f"Bug: {str(bug['id'])} Summary: {bug['summary']}\n")
            # we splite the lines to lists and try to match the words
            matches = 0
            bug_words = list(bug['summary'].split())
            # add c/p and v
            for word in bug_words:
                if pkgsplit(word) is not None:
                     bug_words.append(pkgsplit(word)[0])
                     bug_words.append(cpv_getversion(word))
            print(bug_words)
            print(self.getProperty('error_dict')['words'])
            for word in bug_words:
                # clean the word
                word = finishTitle(stripQuotesAndMore(word))
                for match_word in self.getProperty('error_dict')['words']:
                    if match_word == word:
                        matches = matches + 1
            print(f"Bug: {str(bug['id'])} Matched words: {str(matches)} Summary: {bug['summary']}")
            len_match_words = len(self.getProperty('error_dict')['words']) - self.bug_config['min_match_word']
            print(f"Min words to matchs: {str(len_match_words)}")
            if matches >= len_match_words or matches >= self.bug_config['max_match_word']:
                match = {}
                match['match'] = True
                match['id'] = bug['id']
                match['summary'] = bug['summary']
        yield log.addStdout(f"Line to match: {self.getProperty('error_dict')['title_issue']}\n")
        yield log.addStdout(f"Nice line to match: {self.getProperty('error_dict')['title_nice']}\n")
        yield log.addStdout(f"Words to match: {self.getProperty('error_dict')['words']}\n")
        if match:
            yield log.addStdout('Match bug: YES\n')
            yield log.addStdout(f"Bug: {str(match['id'])} Summary: {match['summary']}\n")
            self.setProperty("bgo", match, 'bgo')
            print(f"Match bug: YES")
            print(f"Bug: {str(match['id'])} Summary: {match['summary']}")
            return match
        yield log.addStdout('Match bug: NO\n')
        return match

    @defer.inlineCallbacks
    def run(self):
        self.gentooci = self.master.namedServices['services'].namedServices['gentooci']
        self.bug_config = self.gentooci.config.project['bug_config']
        cpv = self.getProperty('error_dict')['cpv']
        c = yield catpkgsplit(cpv)[0]
        p = yield catpkgsplit(cpv)[1]
        cp = c + '/' + p
        # search for open bugs
        args = []
        args.append('--skip-auth')
        args.append('search')
        # set limit
        # set date last 30 days
        # search for cp
        args.append(cp)
        print(args)
        buglist = search_bugz(args)
        print(buglist)
        match = self.find_match(buglist)
        #FIXME: set bug id on build in db
        return SUCCESS

class Post(BuildStep):

    name = 'Post bug'
    description = 'Running'
    descriptionSuffix = None
    haltOnFailure = True
    flunkOnFailure = True

    def __init__(self, bug_args, bugs_params, **kwargs):
        self.bug_args = bug_args
        self.bugs_params = bugs_params
        super().__init__(**kwargs)

    @defer.inlineCallbacks
    def run(self):
        #self.gentooci = self.master.namedServices['services'].namedServices['gentooci']
        #bug_config = self.gentooci.config.project['bug_config']
        args = []
        args.append('-u')
        args.append(self.bug_args['user'])
        args.append('-p')
        args.append(self.bug_args['passwd'])
        args.append('post')
        params = {}
        params['product'] = "Gentoo Linux"
        params['component'] = "Current packages"
        params['version'] = "unspecified"
        params['op_sys'] = "Linux"
        params['platform'] = "All"
        params['priority'] = "Normal"
        params['description'] = self.bugs_params['description']
        params['summary'] = self.bugs_params['summary']
        params['assigned_to'] = self.bugs_params['assigned_to']
        #params['cc'] = settings.cc
        params['url'] = self.bugs_params['url']
        bug_info = yield post_bugs(args, params)
        match = {}
        match['match'] = True
        match['id'] = bug_info['id']
        match['summary'] = self.bugs_params['summary']
        self.setProperty("bgo", match, 'bgo')
        print(match)
        self.descriptionDone = f"Bug: {bug_info['id']} submitted"
        #FIXME: update build with bug id
        return SUCCESS

class Modify(BuildStep):
    name = 'Modify'
    description = 'Running'
    descriptionSuffix = None
    haltOnFailure = True
    flunkOnFailure = True

    def __init__(self, bug_args, bug_params, **kwargs):
        self.bug_args = bug_args
        self.bug_params = bug_params
        super().__init__(**kwargs)

    @defer.inlineCallbacks
    def run(self):
        bugid = str(self.getProperty('bgo')['id'])
        args = []
        args.append('-u')
        args.append(self.bug_args['user'])
        args.append('-p')
        args.append(self.bug_args['passwd'])
        args.append('modify')
        args.append(bugid)
        params = {}
        if self.bug_params['comment'] is not None:
            params['comment'] = {}
            params['comment']['body'] = self.bug_params['comment']
        params['ids'] = [bugid]
        print(params)
        bug_info = yield modify_bugs(args, params)
        for bug in bug_info['bugs']:
            changes = bug['changes']
            if not len(changes):
                log_info('Added comment to bug %s' % bug['id'])
            else:
                log_info('Modified the following fields in bug %s' % bug['id'])
                for key in changes:
                    log_info('%-12s: removed %s' %(key, changes[key]['removed']))
                    log_info('%-12s: added %s' %(key, changes[key]['added']))
        self.descriptionDone = f"Modified bug {self.getProperty('bgo')['id']}"
        return SUCCESS

class Attach(BuildStep):
    name = 'Attach'
    description = 'Running'
    descriptionSuffix = None
    haltOnFailure = True
    flunkOnFailure = True

    def __init__(self, bug_args, bug_params, **kwargs):
        self.bug_args = bug_args
        self.bug_params = bug_params
        super().__init__(**kwargs)

    @defer.inlineCallbacks
    def run(self):
        bugid = str(self.getProperty('bgo')['id'])
        args = []
        args.append('-u')
        args.append(self.bug_args['user'])
        args.append('-p')
        args.append(self.bug_args['passwd'])
        args.append('attach')
        args.append(bugid)
        args.append(self.bug_params['filename'])
        summary = os.path.basename(self.bug_params['filename'])

        if not os.path.exists(self.bug_params['filename']):
            raise BugzError('File not found: %s' % self.bug_params['filename'])

        params = {}
        params['ids'] = bugid
        params['file_name'] = summary
        params['summary'] = summary
        if self.bug_params['filename'].endswith('gz'):
            params['content_type'] = 'application/gzip'
        elif self.bug_params['filename'].endswith('bz2'):
            params['content_type'] = 'application/x-bzip'
        elif self.bug_params['filename'].endswith('xz'):
            params['content_type'] = 'application/x-xz'
        else:
            params['content_type'] = 'text/plain'
        #params['comment'] = self.bug_params['comment']
        params['filename'] = self.bug_params['filename']
        print(f"Add params['filename'] to bug {bugid}")
        #params['is_patch'] = is_patch
        bug_info = yield attach_bugs(args, params)
        log_info("'%s' has been attached to bug %s" % (params['filename'], bugid))
        self.descriptionDone = f"{os.path.basename(params['filename'])} has been attached to bug {bugid}"
        return SUCCESS