hgext/bugzilla.py

16218:81a1a00f5738

# bugzilla.py - bugzilla integration for mercurial
#
# Copyright 2006 Vadim Gelfer <[e-mail]>
# Copyright 2011 Jim Hague <[e-mail]>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

'''hooks for integrating with the Bugzilla bug tracker

            if committer.lower() == user.lower():
                return bzuser
        return user

    # Methods to be implemented by access classes.
    def filter_real_bug_ids(self, ids):
        '''remove bug IDs that do not exist in Bugzilla from set.'''
        pass

    def filter_cset_known_bug_ids(self, node, ids):
        '''remove bug IDs where node occurs in comment text from set.'''
        pass

    def add_comment(self, bugid, text, committer):
        '''add comment to bug.

        If possible add the comment as being from the committer of
        the changeset. Otherwise use the default Bugzilla user.
        '''
        pass

    def notify(self, ids, committer):
        '''Force sending of Bugzilla notification emails.'''
        pass

# Bugzilla via direct access to MySQL database.
class bzmysql(bzaccess):
    '''Support for direct MySQL access to Bugzilla.

        ids = self.cursor.fetchall()
        if len(ids) != 1:
            raise util.Abort(_('unknown database schema'))
        return ids[0][0]

    def filter_real_bug_ids(self, ids):
        '''filter not-existing bug ids from set.'''
        self.run('select bug_id from bugs where bug_id in %s' %
                 bzmysql.sql_buglist(ids))
        return set([c[0] for c in self.cursor.fetchall()])

    def filter_cset_known_bug_ids(self, node, ids):
        '''filter bug ids that already refer to this changeset from set.'''

        self.run('''select bug_id from longdescs where
                    bug_id in %s and thetext like "%%%s%%"''' %
                 (bzmysql.sql_buglist(ids), short(node)))
        for (id,) in self.cursor.fetchall():
            self.ui.status(_('bug %d already knows about changeset %s\n') %
                           (id, short(node)))
            ids.discard(id)
        return ids

    def notify(self, ids, committer):
        '''tell bugzilla to send mail.'''

        self.ui.status(_('telling bugzilla to send mail:\n'))
        (user, userid) = self.get_bugzilla_user(committer)
        for id in ids:
            self.ui.status(_('  bug %s\n') % id)
            cmdfmt = self.ui.config('bugzilla', 'notify', self.default_notify)
            bzdir = self.ui.config('bugzilla', 'bzdir', '/var/www/html/bugzilla')
            try:
                # Backwards-compatible with old notify string, which

            except KeyError:
                raise util.Abort(_('cannot find bugzilla user id for %s or %s') %
                                 (user, defaultuser))
        return (user, userid)

    def add_comment(self, bugid, text, committer):
        '''add comment to bug. try adding comment as committer of
        changeset, otherwise as default bugzilla user.'''
        (user, userid) = self.get_bugzilla_user(committer)
        now = time.strftime('%Y-%m-%d %H:%M:%S')
        self.run('''insert into longdescs
                    (bug_id, who, bug_when, thetext)
                    values (%s, %s, %s, %s)''',

        else:
            return cookietransport()

    def get_bug_comments(self, id):
        """Return a string with all comment text for a bug."""
        c = self.bzproxy.Bug.comments(dict(ids=[id]))
        return ''.join([t['text'] for t in c['bugs'][str(id)]['comments']])

    def filter_real_bug_ids(self, ids):
        res = set()
        bugs = self.bzproxy.Bug.get(dict(ids=sorted(ids), permissive=True))
        for bug in bugs['bugs']:
            res.add(bug['id'])
        return res

    def filter_cset_known_bug_ids(self, node, ids):
        for id in sorted(ids):
            if self.get_bug_comments(id).find(short(node)) != -1:
                self.ui.status(_('bug %d already knows about changeset %s\n') %
                               (id, short(node)))
                ids.discard(id)
        return ids

    def add_comment(self, bugid, text, committer):
        self.bzproxy.Bug.add_comment(dict(id=bugid, comment=text))

class bzxmlrpcemail(bzxmlrpc):
    """Read data from Bugzilla via XMLRPC, send updates via email.

    Advantages of sending updates via email:

        msg['To'] = bzemail
        msg['Subject'] = mail.headencode(self.ui, "Bug modification", _charsets)
        sendmail = mail.connect(self.ui)
        sendmail(user, bzemail, msg.as_string())

    def add_comment(self, bugid, text, committer):
        self.send_bug_modify_email(bugid, [], text, committer)

class bugzilla(object):
    # supported versions of bugzilla. different versions have
    # different schemas.

        return getattr(self.bz(), key)

    _bug_re = None
    _split_re = None

    def find_bug_ids(self, ctx):
        '''return set of integer bug IDs from commit comment.

        Extract bug IDs from changeset comments. Filter out any that are
        not known to Bugzilla, and any that already have a reference to
        the given changeset in their comments.
        '''
        if bugzilla._bug_re is None:
            bugzilla._bug_re = re.compile(
                self.ui.config('bugzilla', 'regexp', bugzilla._default_bug_re),
                re.IGNORECASE)
            bugzilla._split_re = re.compile(r'\D+')
        start = 0
        ids = set()
        while True:
            m = bugzilla._bug_re.search(ctx.description(), start)
            if not m:
                break
            start = m.end()
            for id in bugzilla._split_re.split(m.group(1)):
                if not id:
                    continue
                ids.add(int(id))
        if ids:
            ids = self.filter_real_bug_ids(ids)
        if ids:
            ids = self.filter_cset_known_bug_ids(ctx.node(), ids)
        return ids

    def update(self, bugid, ctx):
        '''update bugzilla bug with reference to changeset.'''

        def webroot(root):
            '''strip leading prefix of repo root and turn into
            url-safe path.'''

               bug=str(bugid),
               hgweb=self.ui.config('web', 'baseurl'),
               root=self.repo.root,
               webroot=webroot(self.repo.root))
        data = self.ui.popbuffer()
        self.add_comment(bugid, data, util.email(ctx.user()))

def hook(ui, repo, hooktype, node=None, **kwargs):
    '''add comment to bugzilla for each changeset that refers to a
    bugzilla bug id. only add a comment once per bug, so same change
    seen multiple times does not fill bug with duplicate data.'''

        raise util.Abort(_('hook type %s does not pass a changeset id') %
                         hooktype)
    try:
        bz = bugzilla(ui, repo)
        ctx = repo[node]
        ids = bz.find_bug_ids(ctx)
        if ids:
            for id in ids:
                bz.update(id, ctx)
            bz.notify(ids, util.email(ctx.user()))
    except Exception, e:
        raise util.Abort(_('Bugzilla error: %s') % e)

16221:4fc9fcd991c1

# bugzilla.py - bugzilla integration for mercurial
#
# Copyright 2006 Vadim Gelfer <[e-mail]>
# Copyright 2011-2 Jim Hague <[e-mail]>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.

'''hooks for integrating with the Bugzilla bug tracker

            if committer.lower() == user.lower():
                return bzuser
        return user

    # Methods to be implemented by access classes.
    #
    # 'bugs' is a dict keyed on bug id, where values are a dict holding
    # updates to bug state. Currently no states are recognised, but this
    # will change soon.
    def filter_real_bug_ids(self, bugs):
        '''remove bug IDs that do not exist in Bugzilla from bugs.'''
        pass

    def filter_cset_known_bug_ids(self, node, bugs):
        '''remove bug IDs where node occurs in comment text from bugs.'''
        pass

    def updatebug(self, bugid, newstate, text, committer):
        '''update the specified bug. Add comment text and set new states.

        If possible add the comment as being from the committer of
        the changeset. Otherwise use the default Bugzilla user.
        '''
        pass

    def notify(self, bugs, committer):
        '''Force sending of Bugzilla notification emails.

        Only required if the access method does not trigger notification
        emails automatically.
        '''
        pass

# Bugzilla via direct access to MySQL database.
class bzmysql(bzaccess):
    '''Support for direct MySQL access to Bugzilla.

        ids = self.cursor.fetchall()
        if len(ids) != 1:
            raise util.Abort(_('unknown database schema'))
        return ids[0][0]

    def filter_real_bug_ids(self, bugs):
        '''filter not-existing bugs from set.'''
        self.run('select bug_id from bugs where bug_id in %s' %
                 bzmysql.sql_buglist(bugs.keys()))
        existing = [id for (id,) in self.cursor.fetchall()]
        for id in bugs.keys():
            if id not in existing:
                self.ui.status(_('bug %d does not exist\n') % id)
                del bugs[id]

    def filter_cset_known_bug_ids(self, node, bugs):
        '''filter bug ids that already refer to this changeset from set.'''
        self.run('''select bug_id from longdescs where
                    bug_id in %s and thetext like "%%%s%%"''' %
                 (bzmysql.sql_buglist(bugs.keys()), short(node)))
        for (id,) in self.cursor.fetchall():
            self.ui.status(_('bug %d already knows about changeset %s\n') %
                           (id, short(node)))
            del bugs[id]

    def notify(self, bugs, committer):
        '''tell bugzilla to send mail.'''
        self.ui.status(_('telling bugzilla to send mail:\n'))
        (user, userid) = self.get_bugzilla_user(committer)
        for id in bugs.keys():
            self.ui.status(_('  bug %s\n') % id)
            cmdfmt = self.ui.config('bugzilla', 'notify', self.default_notify)
            bzdir = self.ui.config('bugzilla', 'bzdir', '/var/www/html/bugzilla')
            try:
                # Backwards-compatible with old notify string, which

            except KeyError:
                raise util.Abort(_('cannot find bugzilla user id for %s or %s') %
                                 (user, defaultuser))
        return (user, userid)

    def updatebug(self, bugid, newstate, text, committer):
        '''update bug state with comment text.

        Try adding comment as committer of changeset, otherwise as
        default bugzilla user.'''
        (user, userid) = self.get_bugzilla_user(committer)
        now = time.strftime('%Y-%m-%d %H:%M:%S')
        self.run('''insert into longdescs
                    (bug_id, who, bug_when, thetext)
                    values (%s, %s, %s, %s)''',

        else:
            return cookietransport()

    def get_bug_comments(self, id):
        """Return a string with all comment text for a bug."""
        c = self.bzproxy.Bug.comments(dict(ids=[id], include_fields=['text']))
        return ''.join([t['text'] for t in c['bugs'][str(id)]['comments']])

    def filter_real_bug_ids(self, bugs):
        probe = self.bzproxy.Bug.get(dict(ids=sorted(bugs.keys()),
                                          include_fields=[],
                                          permissive=True))
        for badbug in probe['faults']:
            id = badbug['id']
            self.ui.status(_('bug %d does not exist\n') % id)
            del bugs[id]

    def filter_cset_known_bug_ids(self, node, bugs):
        for id in sorted(bugs.keys()):
            if self.get_bug_comments(id).find(short(node)) != -1:
                self.ui.status(_('bug %d already knows about changeset %s\n') %
                               (id, short(node)))
                del bugs[id]

    def updatebug(self, bugid, newstate, text, committer):
        args = dict(id=bugid, comment=text)
        self.bzproxy.Bug.add_comment(args)

class bzxmlrpcemail(bzxmlrpc):
    """Read data from Bugzilla via XMLRPC, send updates via email.

    Advantages of sending updates via email:

        msg['To'] = bzemail
        msg['Subject'] = mail.headencode(self.ui, "Bug modification", _charsets)
        sendmail = mail.connect(self.ui)
        sendmail(user, bzemail, msg.as_string())

    def updatebug(self, bugid, newstate, text, committer):
        self.send_bug_modify_email(bugid, [], text, committer)

class bugzilla(object):
    # supported versions of bugzilla. different versions have
    # different schemas.

        return getattr(self.bz(), key)

    _bug_re = None
    _split_re = None

    def find_bugs(self, ctx):
        '''return bugs dictionary created from commit comment.

        Extract bug info from changeset comments. Filter out any that are
        not known to Bugzilla, and any that already have a reference to
        the given changeset in their comments.
        '''
        if bugzilla._bug_re is None:
            bugzilla._bug_re = re.compile(
                self.ui.config('bugzilla', 'regexp', bugzilla._default_bug_re),
                re.IGNORECASE)
            bugzilla._split_re = re.compile(r'\D+')
        start = 0
        bugs = {}
        while True:
            m = bugzilla._bug_re.search(ctx.description(), start)
            if not m:
                break
            start = m.end()
            for id in bugzilla._split_re.split(m.group(1)):
                if not id:
                    continue
                bugs[int(id)] = {}
        if bugs:
            self.filter_real_bug_ids(bugs)
        if bugs:
            self.filter_cset_known_bug_ids(ctx.node(), bugs)
        return bugs

    def update(self, bugid, newstate, ctx):
        '''update bugzilla bug with reference to changeset.'''

        def webroot(root):
            '''strip leading prefix of repo root and turn into
            url-safe path.'''

               bug=str(bugid),
               hgweb=self.ui.config('web', 'baseurl'),
               root=self.repo.root,
               webroot=webroot(self.repo.root))
        data = self.ui.popbuffer()
        self.updatebug(bugid, newstate, data, util.email(ctx.user()))

def hook(ui, repo, hooktype, node=None, **kwargs):
    '''add comment to bugzilla for each changeset that refers to a
    bugzilla bug id. only add a comment once per bug, so same change
    seen multiple times does not fill bug with duplicate data.'''

        raise util.Abort(_('hook type %s does not pass a changeset id') %
                         hooktype)
    try:
        bz = bugzilla(ui, repo)
        ctx = repo[node]
        bugs = bz.find_bugs(ctx)
        if bugs:
            for bug in bugs:
                bz.update(bug, bugs[bug], ctx)
            bz.notify(bugs, util.email(ctx.user()))
    except Exception, e:
        raise util.Abort(_('Bugzilla error: %s') % e)